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:

  1. Flux checks for a .pkl file on disk matching the key.
  2. If the file exists, it deserializes the stored value with dill and returns it immediately.
  3. 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 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:

Avoid caching tasks that:

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