Martijn Sanderse
09/05/2025, 12:11 PM"state": {
"id": "019919bf-f823-7153-990f-8779580ab822",
"type": "COMPLETED",
"name": "Completed",
"timestamp": "2025-09-05T12:00:35.363054Z",
"message": null,
"data": {
"storage_key": "/root/.prefect/storage/5419026f36f948c396a057500d24261c",
"expiration": null,
"serializer": {
"type": "json",
"jsonlib": "json",
"object_encoder": "prefect.serializers.prefect_json_object_encoder",
"object_decoder": "prefect.serializers.prefect_json_object_decoder",
"dumps_kwargs": {},
"loads_kwargs": {}
},
"prefect_version": "3.4.14",
"storage_block_id": null
},
"state_details": {
"flow_run_id": "78b4288d-a6a1-493f-b578-b5d257e47a64",
"task_run_id": null,
"child_flow_run_id": null,
"scheduled_time": null,
"cache_key": null,
"cache_expiration": null,
"deferred": null,
"untrackable_result": false,
"pause_timeout": null,
"pause_reschedule": false,
"pause_key": null,
"run_input_keyset": null,
"refresh_cache": null,
"retriable": null,
"transition_id": "1ce9fb5b-3ba7-4e22-ac55-1049ab27fcfb",
"task_parameters_id": null,
"traceparent": null,
"deployment_concurrency_lease_id": null
}
},
How can I retrieve the result using the prefect cli, or a python script?Marvin
09/05/2025, 12:11 PMMarvin
09/05/2025, 12:16 PM/root/.prefect/storage/...
with a JSON serializer.
Here’s how to retrieve it:
CLI
- There isn’t a CLI command that resolves and prints the result payload. prefect flow-run inspect <FLOW_RUN_ID>
will show you the flow run and state metadata (including the storage key), but it won’t load the result content for you.
- You can see the help here: CLI: flow-run commands
- Command you can run:
prefect flow-run inspect 78b4288d-a6a1-493f-b578-b5d257e47a64
- If the storage key points to a local file you can access (e.g., same machine/container), you can read it yourself (see “Direct file read” below).
Python (recommended)
- Given a flow run id, Prefect provides a helper to resolve the state’s result:
import asyncio
from prefect.client.orchestration import get_client
from prefect.states import get_state_result
async def fetch_result(flow_run_id: str):
async with get_client() as client:
flow_run = await client.read_flow_run(flow_run_id)
# Resolve and return the Python object stored as the flow's result
return get_state_result(flow_run.state)
if __name__ == "__main__":
result = asyncio.run(fetch_result("78b4288d-a6a1-493f-b578-b5d257e47a64"))
print("Flow result:", result)
- If you need a task result instead, use read_task_run(task_run_id)
and pass that state to get_state_result
the same way.
Direct file read (when you’re on the same host/container as the worker)
- Since your state points to a local storage key and uses the JSON serializer, you can read and decode it directly:
import json
from prefect.serializers import prefect_json_object_decoder
path = "/root/.prefect/storage/5419026f36f948c396a057500d24261c"
with open(path, "rb") as f:
value = json.loads(f.read(), object_hook=prefect_json_object_decoder)
print("Flow result:", value)
Notes and gotchas
- Access to the file path: /root/.prefect/storage/...
is the worker’s local filesystem path. If you run the retrieval on a different machine/container, you won’t be able to read it unless you use shared/remote result storage (e.g., S3/GCS). The Python get_state_result
helper will read from whatever storage the state references (local, S3, etc.), provided your environment has access and credentials.
- If the result has been cleaned up or storage is inaccessible, get_state_result
will raise an error.
- Docs and references:
- Results API (result storage/serializers)
- prefect.states.get_state_result source
- CLI: flow-run
- REST: read flow run state
If you share where you’re running the retrieval (same node vs. remote) or if you’re using remote result storage, I can tailor the example (e.g., S3/GCS) and ensure the right dependencies are in place.