Hello! I have a question regarding assets with dynamic partitions. How do I define a schedule that would run backfill for them? I can do this easily by running the job in the UI, but the schedule executes a non-partitioned run, which results in an error in the asset function. My current code looks like this:

my_job = define_asset_job(
    "my_job",
    AssetSelection.keys(
        "asset1",
        "asset2"
    ),
    partitions_def=my_dynamic_partitions_def
)

my_schedule = ScheduleDefinition(
    job=my_job,
    cron_schedule="* * * * *",
    default_status=DefaultScheduleStatus.RUNNING
)

Hey - I wrote up an example for this here: https://github.com/dagster-io/dagster/discussions/15532

Thanks for taking the time to create this. I just tested this with Dagster 1.3.9 and it results in the following error:

AttributeError: 'ScheduleEvaluationContext' object has no attribute 'get_partition_keys'

oops - fixed to

partition_keys = partitions_def.get_partition_keys(dynamic_partitions_store=context.instance)

Thanks, that works now 🙂