Configuring behavior based on scheduled run time

This example demonstrates how to use run config to vary the behavior of a run based on its scheduled launch time.

src/<project_name>/defs/assets.py

import dagster as dg


class AssetConfig(dg.Config):
    scheduled_date: str


@dg.asset
def configurable_asset(context: dg.AssetExecutionContext, config: AssetConfig):
    context.log.info(config.scheduled_date)


@dg.schedule(target=configurable_asset, cron_schedule="0 0 * * *")
def configurable_job_schedule(context: dg.ScheduleEvaluationContext):
    scheduled_date = context.scheduled_execution_time.strftime("%Y-%m-%d")
    return dg.RunRequest(
        run_key=None,
        run_config={
            "ops": {
                "configurable_asset": {"config": {"scheduled_date": scheduled_date}}
            }
        },
        tags={"date": scheduled_date},
    )

APIs in this example

• @dg.asset
• AssetExecutionContext
• ScheduleEvaluationContext
• RunRequest