Differential D8684 Diff 40813 examples/docs_snippets/docs_snippets/concepts/partitions_schedules_sensors/schedules/schedule_examples.py

Changeset View

Standalone View

View Options

examples/docs_snippets/docs_snippets/concepts/partitions_schedules_sensors/schedules/schedule_examples.py

Show First 20 Lines • Show All 93 Lines • ▼ Show 20 Lines	return {
}		}
}		}


from dagster import build_schedule_context, validate_run_config		from dagster import build_schedule_context, validate_run_config


def test_my_cron_schedule_with_context():		def test_my_cron_schedule_with_context():
context = build_schedule_context(scheduled_execution_time=datetime.datetime(2020, 1, 1))		context = build_schedule_context(
		scheduled_execution_time=datetime.datetime(2020, 1, 1)
		)
run_config = my_schedule_uses_context(context)		run_config = my_schedule_uses_context(context)
assert validate_run_config(pipeline_for_test, run_config)		assert validate_run_config(pipeline_for_test, run_config)


# end_test_cron_schedule_context		# end_test_cron_schedule_context

# start_hourly_schedule		# start_hourly_schedule


@hourly_schedule(		@hourly_schedule(
pipeline_name="my_pipeline",		pipeline_name="my_pipeline",
start_date=datetime.datetime(2020, 1, 1),		start_date=datetime.datetime(2020, 1, 1),
execution_time=datetime.time(hour=0, minute=25),		execution_time=datetime.time(hour=0, minute=25),
execution_timezone="US/Central",		execution_timezone="US/Central",
)		)
def my_hourly_schedule(date):		def my_hourly_schedule(date):
return {"solids": {"process_data_for_date": {"config": {"date": date.strftime("%Y-%m-%d %H")}}}}		return {
		"solids": {
		"process_data_for_date": {
		"config": {"date": date.strftime("%Y-%m-%d %H")}
		}
		}
		}


# end_hourly_schedule		# end_hourly_schedule

# start_daily_schedule		# start_daily_schedule


@daily_schedule(		@daily_schedule(
pipeline_name="my_pipeline",		pipeline_name="my_pipeline",
start_date=datetime.datetime(2020, 1, 1),		start_date=datetime.datetime(2020, 1, 1),
execution_time=datetime.time(hour=9, minute=0),		execution_time=datetime.time(hour=9, minute=0),
execution_timezone="US/Central",		execution_timezone="US/Central",
)		)
def my_daily_schedule(date):		def my_daily_schedule(date):
return {"solids": {"process_data_for_date": {"config": {"date": date.strftime("%Y-%m-%d")}}}}		return {
		"solids": {
		"process_data_for_date": {
		"config": {"date": date.strftime("%Y-%m-%d")}
		}
		}
		}


# end_daily_schedule		# end_daily_schedule


# start_weekly_schedule		# start_weekly_schedule


@weekly_schedule(		@weekly_schedule(
pipeline_name="my_pipeline",		pipeline_name="my_pipeline",
start_date=datetime.datetime(2020, 1, 1),		start_date=datetime.datetime(2020, 1, 1),
execution_day_of_week=1, # Monday		execution_day_of_week=1, # Monday
execution_timezone="US/Central",		execution_timezone="US/Central",
)		)
def my_weekly_schedule(date):		def my_weekly_schedule(date):
return {"solids": {"process_data_for_date": {"config": {"date": date.strftime("%Y-%m-%d")}}}}		return {
		"solids": {
		"process_data_for_date": {
		"config": {"date": date.strftime("%Y-%m-%d")}
		}
		}
		}


# end_weekly_schedule		# end_weekly_schedule


# start_monthly_schedule		# start_monthly_schedule


@monthly_schedule(		@monthly_schedule(
pipeline_name="my_pipeline",		pipeline_name="my_pipeline",
start_date=datetime.datetime(2020, 1, 1),		start_date=datetime.datetime(2020, 1, 1),
execution_timezone="US/Central",		execution_timezone="US/Central",
execution_day_of_month=15,		execution_day_of_month=15,
execution_time=datetime.time(hour=9, minute=0),		execution_time=datetime.time(hour=9, minute=0),
)		)
def my_monthly_schedule(date):		def my_monthly_schedule(date):
return {"solids": {"process_data_for_date": {"config": {"date": date.strftime("%Y-%m")}}}}		return {
		"solids": {
		"process_data_for_date": {
		"config": {"date": date.strftime("%Y-%m")}
		}
		}
		}


# end_monthly_schedule		# end_monthly_schedule

preset = PresetDefinition(		preset = PresetDefinition(
"test_preset",		"test_preset",
mode="basic",		mode="basic",
run_config={"solids": {"process_data_for_date": {"config": {"date": ""}}}},		run_config={
		"solids": {"process_data_for_date": {"config": {"date": ""}}}
		},
)		)

# start_preset		# start_preset


@daily_schedule(		@daily_schedule(
start_date=datetime.datetime(2020, 1, 1),		start_date=datetime.datetime(2020, 1, 1),
pipeline_name="my_pipeline",		pipeline_name="my_pipeline",
Show All 17 Lines	@daily_schedule(
start_date=datetime.datetime(2020, 1, 1),		start_date=datetime.datetime(2020, 1, 1),
pipeline_name="my_pipeline",		pipeline_name="my_pipeline",
solid_selection=preset.solid_selection,		solid_selection=preset.solid_selection,
mode=preset.mode,		mode=preset.mode,
tags_fn_for_date=lambda _: preset.tags,		tags_fn_for_date=lambda _: preset.tags,
)		)
def my_modified_preset_schedule(date):		def my_modified_preset_schedule(date):
modified_run_config = copy.deepcopy(preset.run_config)		modified_run_config = copy.deepcopy(preset.run_config)
modified_run_config["solids"]["process_data_for_date"]["config"]["date"] = date.strftime(		modified_run_config["solids"]["process_data_for_date"]["config"][
"%Y-%m-%d"		"date"
)		] = date.strftime("%Y-%m-%d")
return modified_run_config		return modified_run_config


# end_modified_preset		# end_modified_preset