https://prefect.io logo
Title
a

Amit Singh

07/13/2020, 3:21 PM
<!here> @Dylan I'm trying to reload a settings module in a flow for each scheduled iteration, but it never loads the changes in settings module. Is that because of some restrictions with the flow, that I might be overlooking
with 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()
d

Dylan

07/13/2020, 3:25 PM
Hey @Amit Singh, Please try to reserve
@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.
a

Amit Singh

07/13/2020, 3:25 PM
Sure @Dylan. My apologies
d

Dylan

07/13/2020, 3:26 PM
As for your question, we work on a support rotation
Whoever’s on for that day will answer you as soon as they can 👍
(My day is Thursdays :p)
a

Amit Singh

07/13/2020, 3:26 PM
👍 thanks
j

Jim Crist-Harif

07/13/2020, 3:33 PM
Your flow building code (the thing in the
with 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.
a

Amit Singh

07/13/2020, 5:03 PM
Hi @Jim Crist-Harif I did try that too, but lo luck. here is the code
@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()
j

Jim Crist-Harif

07/13/2020, 5:26 PM
Since
get_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
Something like:
@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()