DynamicPartition Schedule with Differing Crons

[valentinedwv]

Rewording:
I would like to dynamically add a schedule to a running instance when a partition is added.

I have looked at build_schedule_from_partitioned_job, but

• applies to static and time-partitioned jobs.
• seems to only add schedules at startup.

So that approach does not add a schedule when a partition is dynamically added.

I presently have all the partitions running on the same (@)schedule. These are internationally distributed sources that would like to be harvested at off hours, and at different frequencies. That means a schedule per source, or at least a schedule per unique cron.

-- original
Is it possible to create a different schedules for different partitions?
We have a list of sources where some should be harvested weekly, and others monthly, and others quarterly.

We also want to spread them out to run at night for resources that are global.

source:
- name: source1
   cron:  "0 0 1 * *"
   tz: America/Los_Angeles
- name: source2
   cron:  "0 12 2 * *"
- name: source3
   cron:  "@weekly"
   tz: America/Los_Angeles

Penciled out logic. Trying to figure out how to get a run request from a ScheduleDefinition

sources_partitions_def = DynamicPartitionsDefinition(name="sources_names_active")

@asset(name="sources_details")
def source():
   return source

@asset(name="sources_names_active")
def source_names():
  # read source['name']
   return sources

@job(partitions_def=sources_partitions_def)
def harvest():
  pass

@asset_sensor()
def create_schedules(context):
     sources_details =  context.repository_def.load_asset_value(AssetKey("sources_all"))
     for source_name in sources_names_active:
        #find the source in the asset
        source = next(s for s in sources_details if s["name"] == source_name)
        if source is not None:
            source_cron = source.get('cron', "@monthly")
            source_tz = source.get('tz', "America/Los_Angeles")
            context.instance.schedules.append(
                ScheduleDefinition(
                job=harvest, cron_schedule=source_cron, execution_timezone=source_tz,
                 #   run_key=f"{source}_scheduled"
                )
            )  
     return SensorResult(
            # run_requests=[
            #     RunRequest(partition_key=source
            #            #    , job_name=f"{source}_load"
            #     , run_key=f"{source}_load"
            #                ) for source in new_sources
            # ],
            run_requests=schedules,
            dynamic_partitions_requests=[
                sources_partitions_def.build_add_request(new_sources)
            ],
        )

[valentinedwv]

Is it possible to add a schedule after the system is running?

if I had a sensor, could I use it to add a schedule?

[garethbrickman]

If you need to trigger a job based on both time-based conditions and event-based conditions, you would typically define both a schedule and a sensor for the job. Each would operate independently according to its configuration.

[valentinedwv]

That's what I have now, but then I only have schedule one schedule, or schedules for existing partitions.

Added some more details, above.
When a partition is added, I need to add a schedule to the running instance.

[garethbrickman]

I think your use case may be covered by one or both of these:

• How to write a schedule that launches a single run that fills a range of asset partitions
• How to write a sensor or schedule that requests a run for every partition on every tick

[valentinedwv]

Thank, yes, presently I can run them all at a @Weekly
but want to set the cron based on some per-partition configuration.