https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have started a task run using futur...
# marvin-ai
i
@Marvin I have started a task run using future = task_name.submit(...). I then want to check whether the task run have failed to know if I should proceed with the response from the previous task response or the current one. What's the best approach?
m
✅ thought for 78.9 seconds
The best approach is usually to inspect the future's state before calling 
.result()
. If you call 
future.result()
directly and the task failed, Prefect will raise the task's exception. That's useful sometimes, but if you want to conditionally choose between a previous response and the current one, checking state first is cleaner. Example:
Copy code
python
from prefect import flow, task

@task
def task_name(x):
    if x < 0:
        raise ValueError("bad input")
    return x * 2

@flow
def my_flow():
    previous_response = "fallback"

    future = task_name.submit(-1)

    # Wait for the task run to finish and inspect its state
    state = future.wait()

    if state.is_completed():
        current_response = future.result()
        return current_response
    else:
        return previous_response
A few notes: * 
future.wait()
waits for completion and returns the final 
State
* 
state.is_completed()
means the task succeeded * 
state.is_failed()
means the task raised an exception * You may also want to consider 
state.is_crashed()
or just 
not state.is_completed()
if any non-success should fall back If you specifically want to distinguish failure:
Copy code
python
state = future.wait()

if state.is_failed():
    return previous_response
elif state.is_completed():
    return future.result()
If you want the exception details from a failed run:
Copy code
python
from prefect.states import get_state_exception

state = future.wait()

if state.is_failed():
    exc = get_state_exception(state)
    print(f"Task failed: {exc}")
    return previous_response
Rule of thumb: - Use 
future.wait()
+ 
state.is_completed()
/ 
state.is_failed()
when you want to branch safely - Use 
future.result()
directly when you want the flow to error if the task failed Relevant docs/source: - Prefect tasks - Prefect futures API If you want, I can also show the best pattern for chaining tasks while falling back to the previous successful result.
i
@Marvin sometimes I need to build the task in a way that it might fail (and I want it to show as failed) but task produces the partial data that I would like to use anyway. How do I ensure that both the task gets failed and it still returns the data?
m
✅ thought for 139.2 seconds
Yes — in Prefect 3.x you can mark a task as failed and still attach partial data to its final state. The cleanest way is to return a 
Failed
state with structured data, then consume that state explicitly with 
return_state=True
. Example:
Copy code
python
from prefect import task, flow
from prefect.states import Failed

@task
def build_response():
    partial = {
        "rows_processed": 80,
        "rows_failed": 20,
        "response": {"some": "usable partial output"},
    }

    return Failed(
        message="Task failed, but partial data is available",
        data=partial,
    )

@flow
def my_flow():
    state = build_response(return_state=True)

    if state.is_failed():
        partial_data = state.result(raise_on_failure=False)
        print("Task failed")
        print(partial_data)
        return partial_data
Key behavior: * 
state.is_failed()
is still 
True
* the task run shows as failed * 
state.result(raise_on_failure=False)
gives you the attached data * 
state.result()
with default behavior would raise on failure So for your use case, a good pattern is:
Copy code
python
state = my_task(return_state=True)
data = state.result(raise_on_failure=False)

if state.is_failed():
    # use partial data, maybe fallback/merge
    ...
else:
    # use normal result
    ...
A few important caveats: * If you call 
future.result()
directly on a failed task, Prefect will normally raise the failure * To safely access the partial payload, work with the state and use 
raise_on_failure=False
* It's best to make the returned payload explicit, e.g. a dict/dataclass like: * 
partial_output
* 
error_summary
* 
processed_count
* 
failed_count
A very practical pattern is to always return a structured object in the state's data:
Copy code
python
from prefect.states import Failed

@task
def process_batch(items):
    successes = []
    failures = []

    for item in items:
        try:
            successes.append(item * 2)
        except Exception as exc:
            failures.append({"item": item, "error": str(exc)})

    if failures:
        return Failed(
            message="Some items failed",
            data={
                "successes": successes,
                "failures": failures,
            },
        )

    return {
        "successes": successes,
        "failures": [],
    }
If your goal is downstream logic inside the same flow run, this works well. If your goal is durable recovery across processes / API retrieval / later inspection, I'd recommend also persisting the partial data externally or using result persistence, because plain in-memory state data may not be the best long-term handoff mechanism. Docs: - Tasks - States - Futures If you want, I can show you a full pattern for “failed but usable partial output” with 
.submit()
futures, since that’s slightly different from using 
return_state=True
directly.
i
@Marvin what's the difference between calling the task using task_name.submit() and task_name(return_state=True)
m
✅ thought for 84.2 seconds
They differ in both execution model and what you get back. 
task_name.submit(...)
 - Schedules the task to run asynchronously within a flow - Returns a 
PrefectFuture
- Lets your flow continue immediately, so you can run tasks in parallel - You later use 
.wait()
, 
.state
, or 
.result()
on the future Example:
Copy code
python
future = task_name.submit(x=1)

# flow continues here

state = future.wait()
if state.is_completed():
    value = future.result()
task_name(..., return_state=True)
 - Calls the task directly and waits for it to finish - Returns a 
State
- This is synchronous from the caller’s perspective - Useful when you want the full final state immediately instead of just the result Example:
Copy code
python
state = task_name(x=1, return_state=True)

if state.is_failed():
    partial = state.result(raise_on_failure=False)
So in practice: Use 
.submit()
when: - you want concurrency / parallelism - you want to launch multiple tasks and inspect them later - you're orchestrating work inside a flow Use 
return_state=True
when: - you want to run the task now and immediately inspect success/failure - you care more about the final 
State
than parallelism - you want simpler control flow A simple mental model: - 
task_name.submit(...)
→ “start this task and give me a handle to it” - 
task_name(..., return_state=True)
→ “run this task and give me its final state” One more concrete comparison:
Copy code
python
@flow
def my_flow():
    # async submission
    future = task_name.submit(1)

    # sync execution returning state
    state = task_name(2, return_state=True)
In that example: - the first task may still be running while the second call executes - the second call blocks until complete For your earlier “failed but still use partial data” case: - if using `.submit()`: call 
future.wait()
, then inspect the returned state - if using `return_state=True`: inspect the state directly Example with `.submit()`:
Copy code
python
future = task_name.submit(...)
state = future.wait()

if state.is_failed():
    data = state.result(raise_on_failure=False)
Example with direct call:
Copy code
python
state = task_name(..., return_state=True)

if state.is_failed():
    data = state.result(raise_on_failure=False)
One important note: 
.submit()
is the normal choice for task orchestration inside flows. If you just call 
task_name(...)
, that's more like a direct task invocation, not concurrent submission. Relevant docs: - Prefect tasks - Prefect futures API - Prefect states API If you want, I can give you a recommended pattern for “parallel tasks with partial-data-on-failure” using 
.submit()
specifically.
Open in Slack
PreviousNext
Powered by