States are rich objects that contain information about the status of a particular task run or flow run.

You can learn anything about a task or flow by examining its current state or the history of its states. For example, a state could tell you that a task:

  • is scheduled to make a third run attempt in an hour
  • succeeded and what data it produced
  • was scheduled to run, but later cancelled
  • used the cached result of a previous run instead of re-running
  • failed because it timed out

By manipulating a relatively small number of task states, Prefect flows can harness the complexity that emerges in workflows.

Only runs have states Flows and tasks are templates that describe what a system does; only when we run the system does it also take on a state.

​
State types

States have names and types. State types are canonical, with specific orchestration rules that apply to transitions into and out of each state type. A state’s name is often, but not always, synonymous with its type. For example, a task run that is running for the first time has a state with the name Running and the type RUNNING. However, if the task retries, that same task run will have the name Retrying and the type RUNNING. Each time the task run transitions into the RUNNING state, the same orchestration rules are applied.

These are the terminal state types with no transitions to any other state type:

  • COMPLETED
  • CANCELLED
  • FAILED
  • CRASHED

The full list of states and state types includes:

NameTypeTerminal?Description
ScheduledSCHEDULEDNoThe run will begin at a particular time in the future.
LateSCHEDULEDNoThe run’s scheduled start time has passed, but it has not transitioned to PENDING (15 seconds by default).
AwaitingRetrySCHEDULEDNoThe run did not complete successfully because of a code issue and had remaining retry attempts.
PendingPENDINGNoThe run has been submitted to run, but is waiting on necessary preconditions to be satisfied.
RunningRUNNINGNoThe run code is currently executing.
RetryingRUNNINGNoThe run code is currently executing after previously not completing successfully.
PausedPAUSEDNoThe run code has stopped executing until it receives manual approval to proceed.
CancellingCANCELLINGNoThe infrastructure on which the code was running is being cleaned up.
CancelledCANCELLEDYesThe run did not complete because a user determined that it should not.
CompletedCOMPLETEDYesThe run completed successfully.
FailedFAILEDYesThe run did not complete because of a code issue and had no remaining retry attempts.
CrashedCRASHEDYesThe run did not complete because of an infrastructure issue.

​
Returned values

When calling a task or a flow, there are three types of returned values:

  • Data: A Python object (such as int, str, dict, list, and so on).
  • State: A Prefect object indicating the state of a flow or task run.
  • PrefectFuture: A Prefect object that contains both data and State.

Returning data is the default behavior any time you call your_task().

Returning Prefect State occurs anytime you call your task or flow with the argument return_state=True.

Returning PrefectFuture is achieved by calling your_task.submit().

​
Return data

By default, running a task returns data:

from prefect import flow, task 


@task 
def add_one(x):
    return x + 1

@flow 
def my_flow():
    result = add_one(1) # return int

The same rule applies for a nested flow:

@flow 
def nested_flow():
    return 42 

@flow 
def my_flow():
    result = nested_flow() # return data

​
Return Prefect state

To return a State instead, add return_state=True as a parameter of your task call:

@flow 
def my_flow():
    state = add_one(1, return_state=True) # return State

To get data from a State, call .result().

@flow 
def my_flow():
    state = add_one(1, return_state=True) # return State
    result = state.result() # return int

The same rule applies for a nested flow:

@flow 
def nested_flow():
    return 42 

@flow 
def my_flow():
    state = nested_flow(return_state=True) # return State
    result = state.result() # return int

​
Return a PrefectFuture

To get a PrefectFuture, add .submit() to your task call.

@flow 
def my_flow():
    future = add_one.submit(1) # return PrefectFuture

To get data from a PrefectFuture, call .result().

@flow 
def my_flow():
    future = add_one.submit(1) # return PrefectFuture
    result = future.result() # return data

To get a State from a PrefectFuture, call .wait().

@flow 
def my_flow():
    future = add_one.submit(1) # return PrefectFuture
    state = future.wait() # return State

​
Final state determination

The final state of a 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 nested flows within it.
    • If any task run or nested flow run failed and none were cancelled, then the final flow run state is marked as FAILED.
    • If any task run or nested flow 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 successfully completed.

See the Final state determination section of the Flows documentation for further details and examples.

Was this page helpful?

Manage resultsCreate run artifacts