Jordan Fisher
04/15/2026, 4:03 PMMarvin
04/15/2026, 4:03 PMMarvin
04/15/2026, 4:08 PMprefect.results.ResultStore
- prefect.results.get_default_result_storage() (returns a WritableFileSystem)
- State-based result resolution:
- task_run.state.aresult(...) / flow_run.state.aresult(...)
Examples
1) Read/write results directly (outside any task/flow)
from prefect.results import ResultStore, get_default_result_storage
# Use the default storage (e.g., ~/.prefect/storage or a configured block)
storage = get_default_result_storage()
store = ResultStore(result_storage=storage)
# Write a value
store.write(obj={"status": "ok", "count": 42}, key="my/custom/key.json")
# Check existence (respects expiration if set)
print(store.exists("my/custom/key.json")) # True/False
# Read it back
record = store.read("my/custom/key.json")
print(record.result) # {'status': 'ok', 'count': 42}
print(record.metadata) # includes serializer, expiration, storage_key, etc.
Async equivalents are awrite, aread, and aexists.
2) Retrieve a previous run’s result by run ID (no task/flow context required)
import asyncio
from uuid import UUID
from prefect import get_client
async def get_task_result(task_run_id: UUID):
async with get_client() as client:
tr = await client.read_task_run(task_run_id)
if tr.state:
# This will fetch from the result store the task wrote to
return await tr.state.aresult(raise_on_failure=False)
return None
# asyncio.run(get_task_result(UUID("...")))
Notes and clarifications
- There is no separate “cache store” to browse. Caching works by persisting results according to a cache policy; those persisted results live in the same result storage.
- If you know the storage key (e.g., via a task’s result_storage_key template), you can read it directly with ResultStore.read(key=...).
- If you only know a run ID, fetch the run via the client and call state.aresult() to resolve the stored result.
- Caching requires persisted results: set persist_result=True on tasks (or configure defaults) for caches to be usable outside execution.
API surface you can rely on (verified)
- prefect.results.ResultStore (methods: write, read, exists and async variants)
- prefect.results.get_default_result_storage
- prefect.states.get_state_result and state.aresult(...)
- prefect.get_client().read_task_run(...) / read_flow_run(...)
If you can share how you’re generating/deriving the keys (e.g., a task result_storage_key format or a specific run ID), I can show a concrete snippet tailored to your setup.