Mike Dedys
12/17/2025, 2:13 PMMarvin
12/17/2025, 2:13 PMMarvin
12/17/2025, 2:20 PMfrom prefect import flow, task
@task
def my_task(x):
return x + 1
@flow
def my_flow():
state = my_task(1, return_state=True)
if state.name == "Cached":
print("Used a cached result")
value = state.result()
print(value)
- If you have a task run ID, you can load the result via the Python client:
from uuid import UUID
from prefect.client.orchestration import get_client
async def load_task_result(task_run_id: str):
async with get_client() as client:
task_run = await client.read_task_run(UUID(task_run_id))
state = task_run.state
value = await state.result()
return value
Docs for more detail:
- Caching concepts: Caching
- States and `state.result()`: States
- Result persistence (where results are stored/how they’re loaded): Persist workflow results