How do I write a sensor or schedule that requests a run for every partition on every tick?

[sryza]

I want my schedule (or sensor) to essentially run a backfill on every tick.

Keyword phrases:

• I want to run a daily schedule over all my dynamic partitions
• I want to run an hourly schedule that fills in all my dynamic partitions

[sryza]

Below is an example that launches a separate run for every partition. To launch a single run that covers a range of partitions, refer to #14622.

from dagster import (
    schedule,
    DynamicPartitionsDefinition,
    asset,
    define_asset_job,
    Definitions,
    RunRequest,
)

partitions_def = DynamicPartitionsDefinition(name="fruits")


@asset(partitions_def=partitions_def)
def asset1():
    ...


@schedule(job=define_asset_job("job1", selection=[asset1]), cron_schedule="@daily")
def schedule1(context):
    partition_keys = partitions_def.get_partition_keys(dynamic_partitions_store=context.instance)
    return [
        RunRequest(
            partition_key=partition_key,
            run_key=f"{context.scheduled_execution_time}_{partition_key}"
        )
        for partition_key in partition_keys
    ]


defs = Definitions(assets=[asset1], schedules=[schedule1])

[sryza]

This example is assuming that the dynamic partitions are added elsewhere and this schedule is just requesting runs for them. If you want to add dynamic partitions in your schedule, you would want to put your RunRequests inside a SensorResult, like in the example here: https://docs.dagster.io/concepts/partitions-schedules-sensors/partitions#dynamically-partitioned-assets.

[yashafromrussia]

This was helpful, thank you @sryza just to note, you also have to provide run_key:

@schedule(job=define_asset_job("job1", selection=[asset1]), cron_schedule="@daily")
def schedule1(context):
    partition_keys = partitions_def.get_partition_keys(dynamic_partitions_store=context.instance)
    now = datetime.now(timezone.utc)
    return [RunRequest(partition_key=partition_key, run_key=f"{now}_{partition_key}") for partition_key in partition_keys]

[sryza]

I don't believe that a run_key is required unless you're using an older version of Dagster.

[somiandras]

I get this error on 1.6.1 with very similar code, so run_key does seem to be required.

dagster._check.CheckError: Invariant failed. Description: Schedules that return multiple RunRequests must specify a run_key in each RunRequest

[sryza]

Ah, my bad - I updated the example