Differential D8684 Diff 40955 examples/docs_snippets/docs_snippets/concepts/partitions_schedules_sensors/partition_definition.py
Changeset View
Changeset View
Standalone View
Standalone View
examples/docs_snippets/docs_snippets/concepts/partitions_schedules_sensors/partition_definition.py
| Show First 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | return [ | ||||
| date_partition_set, | date_partition_set, | ||||
| ] | ] | ||||
| # end_repo_include | # end_repo_include | ||||
| def _weekday_run_config_for_partition(partition): | def _weekday_run_config_for_partition(partition): | ||||
| return {"solids": {"process_data_for_date": {"config": {"date": partition.value}}}} | return { | ||||
| "solids": { | |||||
| "process_data_for_date": {"config": {"date": partition.value}} | |||||
| } | |||||
| } | |||||
| # start_manual_partition_schedule | # start_manual_partition_schedule | ||||
| weekday_partition_set = PartitionSetDefinition( | weekday_partition_set = PartitionSetDefinition( | ||||
| name="weekday_partition_set", | name="weekday_partition_set", | ||||
| pipeline_name="my_data_pipeline", | pipeline_name="my_data_pipeline", | ||||
| partition_fn=lambda: [ | partition_fn=lambda: [ | ||||
| Partition("Monday"), | Partition("Monday"), | ||||
| Show All 9 Lines | |||||
| def weekday_partition_selector( | def weekday_partition_selector( | ||||
| ctx: ScheduleEvaluationContext, partition_set: PartitionSetDefinition | ctx: ScheduleEvaluationContext, partition_set: PartitionSetDefinition | ||||
| ) -> Union[Partition, List[Partition]]: | ) -> Union[Partition, List[Partition]]: | ||||
| """Maps a schedule execution time to the corresponding partition or list of partitions that | """Maps a schedule execution time to the corresponding partition or list of partitions that | ||||
| should be executed at that time""" | should be executed at that time""" | ||||
| partitions = partition_set.get_partitions(ctx.scheduled_execution_time) | partitions = partition_set.get_partitions(ctx.scheduled_execution_time) | ||||
| weekday = ctx.scheduled_execution_time.weekday() if ctx.scheduled_execution_time else 0 | weekday = ( | ||||
| ctx.scheduled_execution_time.weekday() | |||||
| if ctx.scheduled_execution_time | |||||
| else 0 | |||||
| ) | |||||
| return partitions[weekday] | return partitions[weekday] | ||||
| my_schedule = weekday_partition_set.create_schedule_definition( | my_schedule = weekday_partition_set.create_schedule_definition( | ||||
| "my_schedule", | "my_schedule", | ||||
| "5 0 * * *", | "5 0 * * *", | ||||
| partition_selector=weekday_partition_selector, | partition_selector=weekday_partition_selector, | ||||
| execution_timezone="US/Eastern", | execution_timezone="US/Eastern", | ||||
| Show All 13 Lines | |||||