https://prefect.io
Join Slack
What is the prefect way of returning data from a flow run that you want to mark as a failure? I’m t...
# ask-community
l
What is the prefect way of returning data from a flow run that you want to mark as a failure? I’m try to use States, but running into issues:
Copy code
from prefect.states import Completed, Failed
from prefect import task, flow

@flow
def flow_that_fails():
    failed_values = ['e', 'f', 'g']
    return Failed(message = 'some flows failed', data = failed_values)

@flow
def my_flow():
    result = flow_that_fails(return_state = True)
    data = result.result(raise_on_failure = False)
    print(data)

my_flow()
gives:
Copy code
TypeError: Unexpected result for failed state: ['e', 'f', 'g'] —— list cannot be resolved into an exception
I’m on prefect 2.8.3
c
Are you trying to return a state or data or both? I think you want a prefect future for both at the task level
Here’s an example provided for the flow that fails:
Copy code
from prefect import flow, task

@task
def my_task():
    raise ValueError()

@flow
def my_flow():
    try:
        my_task()
    except ValueError:
        print("Oh no! The task failed.")

    return True

my_flow()
The final state of the flow is determined by its return value. The following rules apply: • If an exception is raised directly in the flow function, the flow run is marked as failed. • If the flow does not return a value (or returns 
None
), its state is determined by the states of all of the tasks and subflows within it. • If any task run or subflow run failed, then the final flow run state is marked as 
FAILED
. • If any task run was cancelled, then the final flow run state is marked as 
CANCELLED
. • If a flow returns a manually created state, it is used as the state of the final flow run. This allows for manual determination of final state. • If the flow run returns any other object, then it is marked as completed.
l
I'm trying to return both state and data
So the flow fails, and I return a list
c
Failed() just returns a state
l
What is the data field for then?
c
I don’t really understand what you’re trying to do, but if you want both state and data, then it needs to be a future
Copy code
from prefect import flow, task

@task
def always_fails_task():
    raise ValueError("I fail successfully")

@task
def always_succeeds_task():
    print("I'm fail safe!")
    return "success"

@flow
def always_succeeds_flow():
    x = always_fails_task.submit().result(raise_on_failure=False)
    y = always_succeeds_task.submit(wait_for=[x])
    return y

if __name__ == "__main__":
    always_succeeds_flow()
You can have a failure and still return data, but you’re trying to do everything from the flow, and not a task
l
I think the mistake I am making is flows only return state, and not data?
This is a flow of flows example
c
You can return anything - but if you are returning a state specifically, there is no data with it, it’s just a state https://docs.prefect.io/concepts/flows/#return-an-object https://docs.prefect.io/concepts/flows/#return-a-manual-state https://docs.prefect.io/concepts/flows/#return-a-future
generally the practice is to return a future, and check the state if it’s failed or not
z
Re the error you got at
return Failed(message = ‘some flows failed’, data = failed_values)
The issue is that 
data
for 
Failed
states needs to be an exception type
If you really want to return data from a failed task, you’d need to attach it to the exception
e.g.
Copy code
from prefect import flow, task
from prefect.states import Failed


class MyException(Exception):
    def __init__(self, data):
        self.data = data


@flow
def foo():
    try:
        bar()
    except MyException as exc:
        print(exc.data)

    return True

@task
def bar():
    return Failed(message="oh no", data=MyException(data="test"))


foo()
🙌 1
l
Awesome thanks for the info! I really appreciate it
2 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by