<@ULVA73B9P> Is it possible to interact with cache...
# marvin-ai
j
@Marvin Is it possible to interact with cache stores outside the context of a task? Like can we interact with results stores directly without needing to implement ourselves?
m
thought for 259.9 seconds
Short answer: yes. In Prefect 3.x there isn’t a separate “cache store.” Caching is implemented on top of result persistence, and you can interact with the result store directly outside of tasks/flows. What you can do - Read/write results directly via ResultStore (no flow/task context needed) - Check for existence, expiration, and metadata - Retrieve a task/flow result for a specific run by fetching the run state and resolving its result Key APIs - Result store access: -
prefect.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)
Copy code
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)
Copy code
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.