How to get the intermediary (in progress) vs final state of a task_run from a prefect future

Show and Tell
, ,
1

If you want to get a task’s state while its in progress you can use future.get_state() on the other hand if you wan to see the task’s final state you can use future.wait().

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
import time


@task
def example_task():
    time.sleep(3)
    return {"result": "Example result"}


@flow(task_runner=ConcurrentTaskRunner)
def example_flow():
    a = example_task.submit()

    time.sleep(1)

    # get state of task at invocation
    print(a.get_state().type)
    # get value of result upon completion of task
    print(a.result())
    # get final state upon completion of task
    print(a.wait().type)


if __name__ == "__main__":
    example_flow()

Output:

15:40:37.481 | INFO    | prefect.engine - Created flow run 'cute-spoonbill' for flow 'example-flow'
15:40:38.498 | INFO    | Flow run 'cute-spoonbill' - Created task run 'example_task-0' for task 'example_task'
15:40:38.499 | INFO    | Flow run 'cute-spoonbill' - Submitted task run 'example_task-0' for execution.
StateType.RUNNING
15:40:42.254 | INFO    | Task run 'example_task-0' - Finished in state Completed()
{'result': 'Example result'}
StateType.COMPLETED
15:40:42.456 | INFO    | Flow run 'cute-spoonbill' - Finished in state Completed('All states completed.')

Helpful docs on this:

Return a Prefect Future
Retrieving Results