Using resources in schedules

This example demonstrates how to use resources in schedules. To specify a resource dependency, annotate the resource as a parameter to the schedule's function.

note

This article assumes familiarity with resources, code locations and definitions, and schedule testing.

All Dagster definitions, including schedules and resources, must be attached to a Definitions call.

import dagster as dg

from datetime import datetime

class DateFormatter(dg.ConfigurableResource):
    format: str

    def strftime(self, dt: datetime) -> str:
        return dt.strftime(self.format)

@dg.job
def process_data(): ...

@dg.schedule(job=process_data, cron_schedule="* * * * *")
def process_data_schedule(
    context: dg.ScheduleEvaluationContext,
    date_formatter: DateFormatter,
):
    formatted_date = date_formatter.strftime(context.scheduled_execution_time)

    return dg.RunRequest(
        run_key=None,
        tags={"date": formatted_date},
    )

defs = dg.Definitions(
    jobs=[process_data],
    schedules=[process_data_schedule],
    resources={"date_formatter": DateFormatter(format="%Y-%m-%d")},
)

APIs in this guide

Name	Description
@dg.schedule	Decorator that defines a schedule that executes according to a given cron schedule.
ConfigurableResource
@dg.job	The decorator used to define a job.
RunRequest	A class that represents all the information required to launch a single run.
RunConfig
Definitions