Wait for successful execution of previous flow on Prefect

Mefitico · June 23, 2023, 9:43pm

I’ve posted this on SO and even set a bounty there, that will be open until June 30th 2023.

I’m trying to use Prefect 2 to configure the execution of a few tasks on Windows (everything is on my local machine).

I often have problems with some tasks, that I might need to fix manually and re-run it. After this, I want other tasks to wait until a successful execution of the first task and only then start, without me having to execute failed schedules manually. As I’ve made on the code below, task_that_should_wait waits for task_that_fails to finish, regardless of whether it is completed successfully or failed. They are schedules by one flow for each, the reason for using separate flows is because they are scheduled to execute nominally at different times.

from prefect import flow, task
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
    

@task
def task_that_fails():
    raise(Exception("Failed here")) # Imagine this is a bug, that I'll fix manually
    return True

    
@flow
def flow_that_fails():
    task_that_fails()

deploy_that_fails = Deployment.build_from_flow(
    name="task_that_fails",
    flow=flow_that_fails,
    schedule=CronSchedule(cron="36 21 * * MON,TUE,WED,THU,FRI", timezone='US/Central'),
)

deploy_that_fails.apply()
    
@task
def task_that_should_wait():
    print("hey!")
    return True

    
@flow
def flow_that_waits():
    task_that_should_wait(wait_for=['task_that_fails'])

deploy_that_waits = Deployment.build_from_flow(
    name="deploy_that_waits",
    flow=flow_that_waits,
    schedule=CronSchedule(cron="35 21 * * MON,TUE,WED,THU,FRI", timezone='US/Central'),
)

deploy_that_waits.apply()

How can I configure task_that_should_wait to actually wait for the successful completion of task_that_fails? These are props for processes that take very long and have multiple task dependencies, so think that task_that_waits is not necessary run after task_that_fails, so I also need to avoid making them a single task or flow.

Jeremy_Yu · July 7, 2023, 3:16am

I was looking for the same thing and found the .wait_for=().

github.com/PrefectHQ/prefect

Add `wait_for` for flows

opened 05:50PM - 22 Apr 22 UTC

closed 07:06PM - 01 Nov 22 UTC

marvin-robot

enhancement from:slack status:accepted v2 priority:high v1-parity

## User story As a Prefect 2 user, I want to use a flow as a subflow, with a de…pendency on an upstream flow or a task, even though they do not do not exchange data. I want to do this with the `wait_for` keyword argument, just as I can for tasks. ## Opened from the [Prefect Public Slack Community](https://prefect.io/slack) **digangidomenico**: Hi all, I am experimenting with tasks and subflows in prefect 2.0. While in tasks I can use the wait_for keyword to enforce task dependency, the same keyword does not work for subflows. I did not found much on the subflows docs, but it seems that in prefect 1.0 dependency between subflows was possible. Is such a feature available also in prefect 2 ? **michael054**: This just hasn’t been added yet **michael054**: However, since a subflow will block until it is complete, the `wait_for` kwarg is less useful **michael054**: You might as well just wait for all the tasks to finish using `.wait()` before calling the subflow **digangidomenico**: Thanks for the quick reply <@U01CEUST9B5>. If I understand correctly, using wait_for in a task also injects a "logical dependency" not only a temporal ordering. For example, in the following ``` @task def task_a(): pass @task def task_b(): pass @flow def my_flow(): a = task_a() b = task_b(wait_for=[a]) ``` if task_a fails, task_b does not execute, right? I am looking for a similar dependency between subflows or a subflow depending on a task **michael054**: Ah yes, I see your point. You can pass the task futures to the subflow directly but when you do not want to add a data dependency this introduces a problem **michael054**: You can add a `wait_for` argument to your subflow that takes a list of upstream tasks and just not use the data if you don’t need it. We’ll likely support this as a first-class feature in the future. I’ll open an issue for tracking. **digangidomenico**: I think this will work. Many thanks for your help <@U01CEUST9B5> **michael054**: <@ULVA73B9P> open “Add `wait_for` for flows” Original thread can be found [here](https://prefect-community.slack.com/archives/CL09KU1K7/p1650646059038279?thread_ts=1650646059.038279&cid=CL09KU1K7). ## References [Frequently Asked Questions - Enforcing ordering](https://docs.prefect.io/faq/?h=wait_for#how-do-i-enforce-ordering-between-tasks-that-dont-share-data)