Amit Singh
07/13/2020, 3:21 PMwith Flow('ETL Status Cron Flow', schedule=CronSchedule("*/5 * * * *")) as cron_flow:
work_conf = importlib.import_module('settings.work_settings')
importlib.reload(work_conf)
my_settings = getattr(work_conf, 'my_settings')
if(my_settings['active'] == False):
return
else:
print('do something')
state = cron_flow.run()
Dylan
@here
for emergencies that might affect others in the community (Cloud is down, for example) We do our best to be responsive here during business hours and we promise we’ll get to your question as soon as we can.Amit Singh
07/13/2020, 3:25 PMDylan
Amit Singh
07/13/2020, 3:26 PMJim Crist-Harif
07/13/2020, 3:33 PMwith Flow
block) only runs once (on script startup). To get some code to run on each flow run you need to wrap that code in a task
. I suggest adding a task to your flow that reloads the configuration.Amit Singh
07/13/2020, 5:03 PM@task(name='get_work_settings')
def get_work_settings():
work_conf = importlib.import_module('settings.work_settings')
importlib.reload(work_conf)
return getattr(work_conf, 'my_settings')
with Flow('ETL Status Cron Flow', schedule=CronSchedule("*/5 * * * *")) as cron_flow:
my_settings = get_work_settings()
if(my_settings['active'] == False):
return
else:
print('do something')
state = cron_flow.run()
Jim Crist-Harif
07/13/2020, 5:26 PMget_work_settings
is a task, you probably want to use something like case
instead of an if
block. https://docs.prefect.io/core/idioms/conditional.html@task
def is_active(settings):
return bool(settings["active"])
with Flow(...) as flow:
settings = get_work_settings()
with case(is_active(settings), True):
some_task_if_active()