CA Lee
09/16/2021, 3:30 AMCA Lee
09/16/2021, 3:34 AMdef is_holiday():
holidays_list = [...]
today = pd.to_datetime('today').strftime('%Y-%m-%d')
return True if today in holidays else False
And then wrapping every Prefect task in the below if/else logic:
@task
def example_task():
if not is_holiday():
execute_code()
else:
<http://logger.info|logger.info>('Today is holiday.')
Is there anything natively available in Prefect to handle such cases?
Am looking at conditional logic, but not quite sure on implementation detailsCA Lee
09/16/2021, 3:34 AMCA Lee
09/16/2021, 3:55 AMwith case(is_holiday, True)
block will always execute, regardless of whether the condition is True / False:
@task
def is_holiday():
holidays = [...]
today = pd.to_datetime('today').strftime('%Y-%m'%d)
return True if today in holidays else False
from prefect import case, Flow
with Flow(...) as flow:
with case(is_holiday, True):
<http://logger.info|logger.info>('Today is holiday')
with case(is_holiday, False):
execute_code()
CA Lee
09/16/2021, 4:09 AMis_holiday
will run on each trigger of a task run, is this understanding correct ?CA Lee
09/16/2021, 4:13 AMfrom prefect import case, Flow
from prefect.schedules.schedules import Schedule
...
schedule_one = clocks.CronClock("0 15 * * 1-6", parameter_defaults={"report_number": [
"report_one",
"report_two",
"report_three"
]})
schedule_two = clocks.CronClock("0 20 * * 1-6", parameter_defaults={"report_number": [
"report_four,
"report_five",
"report_six"
]})
with Flow(..., schedule=Schedule(clocks=[
schedule_one,
schedule_two
]) as flow:
with case(is_holiday, True):
<http://logger.info|logger.info>('Today is holiday')
with case(is_holiday, False):
execute_code()
This means that the case function is_holiday
would be executed 6 times?nicholas
CA Lee
09/16/2021, 4:29 AMnicholas
CA Lee
09/16/2021, 5:31 AMfrom prefect import Flow
from prefect.schedules import filters
from prefect.schedules.schedules import Schedule
def holidays():
# List of datetime objects
holidays = [...]
# Return a list of filters
[return filters.on_date(month=date.month, day=date.day) for date in holidays]
with Flow(..., schedule=Schedule(clocks=[
schedule_one,
schedule_two],
# Pass in list of filters into Schedule
not_filters=holidays()
) as flow:
...