I am stuck on how to use the new Definitions API to include a schedule definition for a partitioned asset to be rematerialized daily.

ScheduleDefinition(
    job=define_asset_job(
        "partitioned_assets_job",
        selection=AssetSelection.all(),
        partitions_def=StaticPartitionsDefinition(["red", "blue"]),
    ),
    cron_schedule="45 0 * * *",
    execution_timezone="America/Chicago",
),

Hi @marcos for static partitioned jobs, you’ll need to write out the schedule using the

schedule

decorator so that you can specify which partitions should be materialized yours would look something like this

my_job = define_asset_job(
        "partitioned_assets_job",
        selection=AssetSelection.all(),
        partitions_def=StaticPartitionsDefinition(["red", "blue"]),
    )

@schedule(cron_schedule="45 0 * * *", job=my_job)
def red_schedule():
    request = my_job.run_request_for_partition(
        partition_key="red", run_key=None
    )
    yield request

this docs section goes over how to set this up https://docs.dagster.io/concepts/partitions-schedules-sensors/schedules#static-partitioned-jobs

Ahhh I had read that section so many times, but makes it land. Thank you @jamie!