Caching task results
Skip redundant computation by caching task outputs to disk, and understand how cache keys are derived, where results are stored, and when to bust the cache.
Set cache=True on any task whose inputs deterministically produce the same output. Flux stores the result to a .pkl file the first time the task runs and returns the stored value on every subsequent call — within the same execution and across executions — without invoking the function again.
Enabling the cache
Pass cache=True to @task.with_options:
from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow
@task.with_options(cache=True)
async def expensive_lookup(key: str) -> str:
# Slow or costly operation — a remote API call, a DB query, etc.
return f"result-for-{key}"
@workflow
async def deduplicate(ctx: ExecutionContext[str]):
first = await expensive_lookup(ctx.input)
second = await expensive_lookup(ctx.input) # returned from cache
return {"first": first, "second": second}
Run it and both results are identical, but the function executes only once per unique set of inputs.
What gets cached
The cache captures the task’s return value after its first successful execution. On subsequent calls that resolve to the same cache key:
- Flux checks for a
.pklfile on disk matching the key. - If the file exists, it deserializes the stored value with
dilland returns it immediately. - If the file does not exist, the task function runs normally, and Flux writes the result to disk before returning.
The cache covers the function’s return value only. Side effects inside the function body — sending emails, writing to a database, calling a payment API — are not rolled back or suppressed when a cache hit occurs. Design cached tasks so their side effects are safe to skip on repeated calls, or keep side effects in separate, non-cached tasks.
Cache key derivation
The cache key is the same string as the task’s internal ID, which Flux derives from three components:
- The task’s name — by default the function name; overridden with
name=inwith_options. - The task’s call arguments — all positional and keyword arguments, converted to a hashable representation via
make_hashableand then Python’s built-inhash(). - Both forms of the arguments —
make_hashableis applied to the bound argument dict and to the rawargstuple separately, then combined.
The resulting string takes the form {task_name}_{abs_hash} and maps to a file named {task_name}_{abs_hash}.pkl.
make_hashable handles nested containers (lists become tuples, sets become frozensets, dicts are sorted by key) and common types. Objects that are not natively hashable fall back to their str() representation. Pydantic models, dataclasses, and most primitives hash stably across calls in the same Python process, but beware of objects whose __str__ is non-deterministic (such as objects that include memory addresses in their representation).
Where results are stored
Cached values land in the directory {home}/.cache/, where home defaults to .flux (relative to the working directory when the process starts). Each cached value is a separate file:
.flux/
└── .cache/
├── expensive_lookup_8389290254435026496.pkl
└── expensive_lookup_3287832502943446809.pkl
The home and cache_path values are configurable in flux.toml:
[flux]
home = ".flux"
cache_path = ".cache"
Or via environment variables:
FLUX_HOME=.flux FLUX_CACHE_PATH=.cache
Cross-execution caching
Unlike Flux’s execution-log replay (which skips re-running tasks that already completed in the same execution), the cache=True store persists across executions. Run the same workflow twice with the same input and the second run will find the .pkl files written by the first:
@task.with_options(cache=True)
async def compute(n: int) -> int:
return n * 2
@workflow
async def double(ctx: ExecutionContext[int]):
return await compute(ctx.input)
# First run: compute executes, writes .flux/.cache/compute_<hash>.pkl
r1 = double.run(21) # output: 42
# Second run: Flux finds the .pkl file, skips compute entirely
r2 = double.run(21) # output: 42, compute never called
cache=True is a coarse memoisation layer suited to tasks whose results are stable across workflow runs: reference data lookups, model weights, configuration fetches, and similar read-heavy operations.
Cache invalidation
Flux provides no built-in mechanism to expire or invalidate cache entries. The supported approaches are:
Delete the cache directory. Remove .flux/.cache/ (or the configured equivalent) to force all cached tasks to re-execute on the next run:
rm -rf .flux/.cache/
Delete individual .pkl files. If you know which task and arguments produced a stale entry, locate the file by its name ({task_name}_{hash}.pkl) and delete it. The task will re-execute and write a fresh file.
Disable caching temporarily. Set cache=False in a with_options override to bypass the cache for a specific invocation without deleting the stored files.
Rename the task. Changing the task’s name (via @task.with_options(name="new-name", cache=True)) produces a different cache key, effectively making old cache files invisible to the new name.
When to use caching
cache=True works well for tasks that are:
- Pure (same inputs always produce the same output)
- Expensive relative to a disk read (network calls, heavy computation, large data loads)
- Stable across workflow runs (the data they return does not change between executions)
Avoid caching tasks that:
- Have meaningful side effects that must happen on every call
- Return values that change over time (current timestamps, live inventory counts)
- Operate on arguments whose hash is not stable across processes (see the
PYTHONHASHSEEDnote above) - Produce large objects — each cached value is a full
dillpickle on disk, with no size limit or eviction policy
For finer control — TTLs, versioning, shared distributed state — manage the cache inside the task body using a shared store (Redis, S3, a database) and leave cache=False.
Combining cache with other options
cache=True composes with the rest of the with_options surface. Pair it with retries so a transient failure does not prevent the result from being cached on the next successful attempt:
@task.with_options(
cache=True,
retry_max_attempts=3,
retry_delay=2,
retry_backoff=2,
)
async def fetch_reference_data(endpoint: str) -> dict:
...
The cache check runs before any retry logic: if a .pkl file exists for the key, Flux returns it without attempting the function or any retries. If the file does not exist, the normal retry cycle applies, and a successful result (on any attempt) is written to the cache.
What’s next
- Reference: @task options — complete parameter reference for the
taskclass andwith_options.