https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> for Prefect 3 *How do I return a Pref...
# ask-marvin
t
@Marvin for Prefect 3 How do I return a Prefect 
Completed
state with data from a task? I have a Prefect task that returns 
Failed()
when an exception occurs, so the flow does not crash. On the happy path, I want the task to return data. Ideally, I would like to return something like:
Copy code
return Completed(data=my_data)
instead of just:
Copy code
return my_data
That way, the task signature can explicitly return a 
State
. But I cannot find how to attach data to 
Completed()
. Is this possible in Prefect? What is the correct pattern here?
m
✅ thought for 143.9 seconds
Yes — in Prefect 3.x, you can return 
Completed(data=my_data)
from a task. That said, the usual pattern is: - return plain Python data on success - raise an exception on failure - only return an explicit 
State
when you intentionally want to control the task state yourself So both of these are valid:
Copy code
python
@task
def my_task():
    return {"x": 1}
and
Copy code
python
@task
def my_task():
    return Completed(data={"x": 1})
In both cases, the task ends in 
Completed
. The first is just simpler. Important detail: if you return an explicit state, Prefect uses that state instead of inferring one from the return value. Example:
Copy code
python
from prefect import task
from prefect.states import Completed, Failed

@task
def my_task(should_fail: bool):
    try:
        if should_fail:
            raise ValueError("boom")

        my_data = {"status": "ok", "value": 42}
        return Completed(data=my_data, message="Task succeeded")

    except Exception as exc:
        return Failed(data={"error": str(exc)}, message="Task handled failure")
A few notes on the correct pattern: If your goal is “don’t crash the flow on expected errors”, returning 
Failed()
is usually not the best fit, because the task is still marked failed. Better options are: 1. Return normal data that encodes success/failure
Copy code
python
@task
def my_task():
    try:
        return {"ok": True, "data": do_work()}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}
2. Return 
Completed(data=...)
even for handled/expected error cases
Copy code
python
@task
def my_task():
    try:
        return Completed(data={"ok": True, "data": do_work()})
    except Exception as exc:
        return Completed(data={"ok": False, "error": str(exc)})
This keeps the task in a 
Completed
state while still carrying structured output. So the guidance is: - Simple success path: 
return my_data
- Need explicit state control: 
return Completed(data=my_data)
- Handled business error that should not count as orchestration failure: return a completed result payload, not 
Failed()
- Actual task failure: raise an exception or return 
Failed(...)
Docs / references: - Prefect states - Prefect tasks If you want, I can also show the best type-hinted pattern for a task that always returns either success-data or handled-error-data without using 
State
in the function signature.
t
how to access the data wrapped in Completed
@Marvin how to access the data wrapped in 
Completed
m
✅ thought for 7.7 seconds
If a task returns 
Completed(data=...)
, you usually access the wrapped data via the task run’s 
.result()
. Example:
Copy code
python
from prefect import flow, task
from prefect.states import Completed

@task
def my_task():
    return Completed(data={"x": 1, "y": 2})

@flow
def my_flow():
    future = my_task.submit()
    result = future.result()
    print(result)

my_flow()
result
will be:
Copy code
python
{"x": 1, "y": 2}
If you're calling the task directly inside a flow:
Copy code
python
@flow
def my_flow():
    result = my_task()
    print(result)
you should also get the underlying data, not the 
State
object. If you specifically want the state object and then inspect its data, use `return_state=True`:
Copy code
python
from prefect import flow, task
from prefect.states import Completed

@task
def my_task():
    return Completed(data={"x": 1})

@flow
def my_flow():
    state = my_task(return_state=True)
    print(state)
    print(state.result())

my_flow()
Key distinction: - 
future.result()
→ gives you the wrapped data - 
state.result()
→ gives you the wrapped data - the 
State
object itself contains Prefect metadata, but 
.result()
is the normal way to unwrap the payload So in short: use 
.result()
. Relevant docs: - States - Tasks If you want, I can also show: - how to inspect 
state.data
directly - how this behaves for 
Failed(data=...)
- the difference between 
.submit()
, direct calls, and 
return_state=True
t
@Marvin I return 
Failed
instead of throwing exception because I don't want the exception to bubble up to fail the Flow, also I want to show the task as failed (red) in the UI, so I return the Failed state. Is there a better way
m
✅ thought for 96.5 seconds
Yes — in Prefect 3.x, returning 
Failed()
from the task is actually a reasonable pattern for your use case: - the task shows as failed/red in the UI - the flow does not automatically fail just because a task failed - you keep control over whether the flow should fail or continue That behavior is a core Prefect 3 change: task failure and flow failure are decoupled. So if your goal is exactly: - task should be red - flow should continue / not crash then this is fine:
Copy code
python
from prefect import task
from prefect.states import Failed, Completed

@task
def do_work():
    try:
        data = {"value": 42}
        return Completed(data=data)
    except Exception as exc:
        return Failed(message=str(exc))
However, there are two better-structured variants depending on what you need downstream. --- Best pattern if you want red task + structured failure data Return 
Failed(data=...)
on error and 
Completed(data=...)
on success:
Copy code
python
from prefect import task
from prefect.states import Completed, Failed

@task
def do_work(x: int):
    try:
        if x < 0:
            raise ValueError("x must be non-negative")

        return Completed(data={"ok": True, "value": x * 2})

    except Exception as exc:
        return Failed(
            message="Task failed but flow may continue",
            data={"ok": False, "error": str(exc), "input": x},
        )
Then in the flow, if you want to inspect the state explicitly:
Copy code
python
from prefect import flow

@flow
def my_flow():
    state = do_work(-1, return_state=True)

    if state.is_failed():
        error_info = state.result(raise_on_failure=False)
        print("task failed:", error_info)
        return "continuing anyway"

    result = state.result()
    print("task succeeded:", result)
A useful detail here: - 
state.result()
on a failed state will normally raise - 
state.result(raise_on_failure=False)
gives you the underlying payload/exception without re-raising --- When this is better than raising exceptions Returning 
Failed()
is often better than raising if: - the failure is an expected business outcome - you want the UI to show failure - you do not want control flow to be interrupted in the flow That sounds like your case. --- When raising is better Raising an exception is better if: - this is a true unexpected error - you want retries to behave in the standard way - you want the flow to decide failure via normal Python 
try/except
Though note: in Prefect 3, even raised task failures do not automatically fail the flow unless that failure is surfaced/returned/awaited in a way that propagates. --- One caveat If downstream tasks consume the failed task’s output, failed upstream states can block normal resolution unless you explicitly handle state/future behavior. So if you need downstream logic to continue based on a failed task, the cleanest approach is often: - call task with 
return_state=True
- inspect 
state.is_failed()
- branch manually in the flow Example:
Copy code
python
from prefect import flow, task
from prefect.states import Completed, Failed

@task
def extract(flag: bool):
    if flag:
        return Failed(data={"reason": "bad input"})
    return Completed(data={"rows": [1, 2, 3]})

@task
def fallback():
    return "used fallback"

@flow
def pipeline(flag: bool):
    state = extract(flag, return_state=True)

    if state.is_failed():
        info = state.result(raise_on_failure=False)
        print("extract failed:", info)
        return fallback()

    data = state.result()
    return data
--- Bottom line For your stated goal, no, there isn’t necessarily a better way than returning 
Failed()
. That is a good Prefect 3 pattern when you want: - failed task state in UI - no exception bubbling - flow remains under your control My recommendation: - use 
Completed(data=...)
on success - use 
Failed(message=..., data=...)
on handled failure
- in the flow, use 
return_state=True
if you need to inspect/branch on that result safely Relevant docs: - Prefect states - Manage states If you want, I can show a fully typed example of a task that returns `Completed`/`Failed` plus a flow that branches cleanly on the returned state.
t
@Marvin where is the documentation about the data= argument in 
Completed()
and 
Failed()
m
✅ thought for 5.9 seconds
It’s not very prominently documented in the narrative docs. The clearest places are: 1. The API reference for state constructors / state objects - `prefect.states` API reference That page documents 
Completed()
/ 
Failed()
and shows they accept 
**kwargs
, which are passed into the underlying 
State
, including 
data
. 2. The 
State
schema/reference - State object reference The important bit is that 
State
has a 
data
field. 
Completed()
and 
Failed()
are helper constructors that build a 
State
of the corresponding type, so 
data=
works because it is a valid 
State
field. Conceptually, it is like this:
Copy code
python
Completed(message="ok", data=my_data)
Failed(message="bad", data=my_error_payload)
because those helpers ultimately create a 
State(...)
with that 
data
. 3. Source-backed state docs - States This explains return-value behavior and explicit returned states, though it does not, as far as I can tell, emphasize 
data=
as clearly as the API/source references do. So the short answer is: - narrative docs: mention explicit returned states, but not 
data=
very clearly - API reference/source: that’s where 
data=
is really discoverable If you want, I can point you to the exact 
Completed
/ 
Failed
implementation lines in source and explain why 
data=
is accepted.
2 Views
Open in Slack
PreviousNext
Powered by