Great question. Here are practical best practices for when (and how) to cache tasks in Prefect 3.x: When to cache - Cache tasks that are: - Expensive to compute (API/db calls, large transformations, ML inference) - Deterministic and side-effect free - Reused across runs (e.g., same date partition or id) - Avoid caching tasks that: - Cause side effects (sending emails, writing to external systems) - Depend on rapidly changing external state - Produce non-deterministic outputs (randomness, “now()” timestamps) unless your cache key ignores the changing bits Recommended cache policies - For cross-run reuse: use

cache_policy=INPUTS

(most common) - For strict one-run reuse: rely on the default (Prefect’s

DEFAULT

combines inputs + code + run id) - To invalidate on code changes: combine with

TASK_SOURCE

(e.g.,

INPUTS + TASK_SOURCE

) - If your inputs include noisy fields (timestamps, flags), define a custom

cache_key_fn

that ignores them Set a reasonable TTL - Align

cache_expiration

with data volatility and SLAs: - API responses that change slowly: hours to a day - Partitioned batch data (e.g., date): 1–7 days - Real-time or frequently changing data: short TTL or no cache Persist results and choose storage wisely - Caching requires persisted results; set

persist_result=True

- Use remote result storage (S3/GCS/Azure block) when you have multiple workers/machines so cache hits are shared - For very large results, set

cache_result_in_memory=False

to reduce memory pressure Design robust cache keys - Prefer

INPUTS

for simplicity; use a custom

cache_key_fn

when: - Inputs aren’t serializable - You need to ignore specific parameters (e.g.,

debug

, timestamps) - Keep cache keys compact; if keys get too large for your storage, generate a short, stable key Concurrency and stampede control - If many runs may compute the same key at the same time, consider stricter isolation/locking (e.g., Redis lock manager) to avoid duplicate work - Otherwise, the default behavior favors throughput and may run duplicates concurrently Retries interaction - Cached results bypass retries on subsequent calls - Let flaky tasks cache only after a successful run; don’t use caching to mask intermittent failures Refreshing and invalidation - Use

refresh_cache=True

on specific calls (or the global setting) when you need to rebuild a cache - Including

TASK_SOURCE

in the policy auto-invalidates when code changes Transactions for multi-step consistency - If multiple tasks must “all cache or none,” wrap them in a transaction to avoid partial cached states Starter patterns - Cross-run caching with TTL:

from datetime import timedelta
from prefect import task

@task(cache_policy=INPUTS, cache_expiration=timedelta(hours=4), persist_result=True)
def fetch_partitioned_data(date: str):
    return expensive_fetch(date)

- Custom key that ignores a noisy flag:

from prefect import task

def cache_key_ignore_debug(context, params):
    return f"report:{params['date']}"

@task(cache_key_fn=cache_key_ignore_debug, persist_result=True)
def build_report(date: str, debug: bool = False):
    return generate(date, debug)

- Distributed cache via remote result storage:

@task(
    cache_policy=INPUTS,
    persist_result=True,
    result_storage="s3-bucket/prod-cache",  # saved block name
)
def cached_transform(x: int):
    return transform(x)

Docs to dig deeper - Caching concepts - How to cache workflow steps - Persisting results If you can share what you’re trying to cache (e.g., API calls vs. transforms), how often inputs repeat, and where your flows run (local vs. distributed), I can recommend a concrete policy/TTL and storage setup.