Im running a flow using VertexRun - It has multiple steps and it runs perfectly.
Once a week at a specific time I’d like this flow to invoke another flow that will train a model.
I used this code -
if pendulum.today('America/New_York').weekday() == 2: # Monday is 0 so Wednesday is 2
x_flow = create_flow_run(flow_name=PREFECT_TRAIN_FLOW_NAME, project_name=PREFECT_TRAIN_PROJECT_NAME)
wait_for_flow_a = wait_for_flow_run(x_flow, raise_final_state=True)
This code is of course inside my “with Flow….” code block.
Now when running this code alone (in a dummy flow) - it works and x_flow gets invoked.
But when running this code in my real flow, after some other tasks - nothing happens.
I cannot even see the task of “wait_for_flow” in prefect cloud(flow -> tasks tab) - seems like its getting ignored.
What am I missing here?
8 months ago
Conditional logic with if/else statements should not be used in the Flow block in Prefect 1.0. This will change in Orion, but for now you need to use the case task, because the Flow block only defines a DAG, which gets execute when flow run is created.
btw, you can entirely avoid this if/else block if you attach this schedule:
from prefect.schedules import CronSchedule
schedule = CronSchedule(
cron="0 9 * * WED", start_date=pendulum.datetime(2022, 1, 12, 0, 0, tz="America/New_York")
for sched in schedule.next(10):
it will run your flow every Wednesday your time zone at 9 AM (just example time)
8 months ago
Yeah its not the case here - this flow is running every day and its calling another flow just on Wednesday.
I will use case tasks for that, thanks!!