@Marvin I have a flow that looks like this:

@flow
def test_flow():
    t0_future = t0.submit()
    t1_future = t1.submit(wait_for=[t0_future])
    t2_future = t2.submit(wait_for=[t1_future])
    t3_future = t3.submit(wait_for=[t2_future])

t1 raises an error. After the flow is done, t0 has state completed, t1 has state failed, and t2 and t3 have state NotReady. Is there a way to make the upstream t1's failure propagate to t2 and t3 so that they end in a failed state as well?

In Prefect 3.x, task failures do not automatically propagate to downstream tasks unless explicitly handled. By default, if a task fails, its downstream tasks that depend on it will remain in a "NotReady" state, as you've observed. To propagate the failure, you can use the

allow_failure

utility from Prefect's top-level API to control the behavior of task dependencies. Here's a brief example of how you can modify your flow to propagate the failure:

python
from prefect import flow, task, allow_failure

@task
def t0():
    pass

@task
def t1():
    raise Exception("Task t1 failed")

@task
def t2():
    pass

@task
def t3():
    pass

@flow
def test_flow():
    t0_future = t0.submit()
    t1_future = t1.submit(wait_for=[t0_future])
    t2_future = t2.submit(wait_for=[allow_failure(t1_future)])
    t3_future = t3.submit(wait_for=[allow_failure(t2_future)])

test_flow()

In this setup,

allow_failure

allows the tasks

t2

and

t3

to run even if their upstream dependencies fail, but you can handle the failure logic within those tasks as needed. This approach gives you control over how failures should be managed and propagated. For more details, you can explore the Prefect API documentation and examples: - Prefect API Overview If you need further customization or have specific requirements, feel free to ask!

@Marvin what if I want t3 to run if all of t0, t1, and t2 complete succesfully, or if any one of them fails?

To execute

t3

conditionally based on the completion or failure of

t0

,

t1

, and

t2

, you can utilize Prefect's state management capabilities. Specifically, you can check the state of these tasks and decide whether to run

t3

. Here's a conceptual approach using Prefect's state handling:

python
from prefect import flow, task, get_run_logger

@task
def t0():
    pass

@task
def t1():
    pass  # or raise an exception to simulate failure

@task
def t2():
    pass

@task
def t3():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Running t3")

@flow
def test_flow():
    t0_future = t0.submit()
    t1_future = t1.submit(wait_for=[t0_future])
    t2_future = t2.submit(wait_for=[t1_future])

    # Wait for tasks to complete and check their states
    t0_state = t0_future.wait()
    t1_state = t1_future.wait()
    t2_state = t2_future.wait()

    # Conditional logic to run t3 if all complete or any fails
    if (t0_state.is_completed() and t1_state.is_completed() and t2_state.is_completed()) or \
       (t0_state.is_failed() or t1_state.is_failed() or t2_state.is_failed()):
        t3.submit()

test_flow()

In this setup: - We submit tasks

t0

,

t1

, and

t2

. - We wait for each task to complete and check their states. - We conditionally execute

t3

based on the state of the upstream tasks using

is_completed()

and

is_failed()

methods. You can adjust the logic inside the conditional statement to suit your specific requirements. For more details, you can refer to the Prefect documentation on state management.

@Marvin how can I propagate failures in upstream tasks to downstream tasks manually?

To manually propagate failures from upstream tasks to downstream tasks in Prefect 3.x, you can explicitly check the state of the upstream tasks and set the downstream tasks to a failed state if any upstream task has failed. Here's a conceptual approach:

python
from prefect import flow, task, get_run_logger
from prefect.states import Failed

@task
def t0():
    pass

@task
def t1():
    raise Exception("Task t1 failed")

@task
def t2():
    pass

@task
def t3(upstream_states):
    logger = get_run_logger()
    if any(state.is_failed() for state in upstream_states):
        raise Exception("Upstream task failed, propagating failure to t3")
    <http://logger.info|logger.info>("Running t3")

@flow
def test_flow():
    t0_future = t0.submit()
    t1_future = t1.submit(wait_for=[t0_future])
    t2_future = t2.submit(wait_for=[t1_future])

    # Wait for tasks to complete and check their states
    t0_state = t0_future.wait()
    t1_state = t1_future.wait()
    t2_state = t2_future.wait()

    # Check if any upstream task has failed and propagate failure
    try:
        t3.submit(upstream_states=[t0_state, t1_state, t2_state])
    except Exception as e:
        logger = get_run_logger()
        logger.error(f"Task t3 failed due to upstream failure: {e}")

test_flow()

In this setup: - Each task's state is checked after completion. - If any upstream task has failed, the downstream task

t3

raises an exception to propagate the failure. - The

t3

task explicitly checks the states of its upstream dependencies and decides whether to proceed or fail. This approach allows you to control the failure propagation logic manually. For more detailed guidance, you can explore the Prefect documentation on state management and transactions.

@Marvin is it possible to change the state of a prefect future?

In Prefect 3.x, you can change the state of a

PrefectFuture

by interacting with the underlying task run state. While the

PrefectFuture

itself doesn't directly allow state modification, you can use Prefect's state management functions to achieve this. Here's a conceptual approach: 1. Retrieve the Task Run: Access the task run associated with the future. 2. Set a New State: Use Prefect's state management to set a new state for the task run. Here's an example of how you might do this:

python
from prefect import flow, task, get_client
from prefect.states import Failed
import asyncio

@task
def example_task():
    return "Hello, Prefect!"

@flow
async def example_flow():
    future = example_task.submit()

    # Wait for the task to complete
    await future.wait()

    # Retrieve the task run ID
    task_run_id = future.task_run_id

    # Get the client and set a new state
    async with get_client() as client:
        # Set the task run to a failed state
        await client.set_task_run_state(task_run_id, state=Failed())

example_flow()

In this setup: - The

example_task

is submitted and its future is awaited. - The task run ID is retrieved from the future. - The Prefect client is used to set a new state for the task run, such as

Failed

. This method allows you to programmatically change the state of a task run, effectively propagating failures or altering execution flow as needed. For more detailed information, you can explore Prefect's state management documentation.