Dekel R

    Dekel R

    8 months ago
    Hey all, 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? Thanks
    Anna Geller

    Anna Geller

    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:
    import pendulum
    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):
        print(sched)
    it will run your flow every Wednesday your time zone at 9 AM (just example time)
    Dekel R

    Dekel R

    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!!