In Prefect 3.x, use the

refresh_cache

option on `@task`: -

refresh_cache=True

— always recompute and overwrite the cached result for the task’s cache key (i.e., “refresh”/effectively invalidate for this run). -

refresh_cache=False

— if a valid cached result exists, return it and skip execution; otherwise compute and write to cache. -

refresh_cache=None

(default) — inherit the global setting (env var `PREFECT_TASKS_REFRESH_CACHE`; default is False). You can set it per-run by overriding the task with `with_options`:

from datetime import timedelta
from prefect import task, flow

@task(cache_policy=None, cache_expiration=timedelta(hours=1))
def expensive(x):
    return x + 1

@flow
def my_flow(x: int, refresh: bool = False):
    t = expensive.with_options(refresh_cache=refresh)
    return t(x)

Notes: -

refresh_cache

controls whether Prefect skips to a cached result or forces a recompute and updates the cache; it does not delete cache entries globally. - It works alongside `cache_policy`/`cache_key_fn` (how the key is computed) and

cache_expiration

(how long it’s valid). Docs: - Caching concepts - How-to: Cache workflow steps - Settings reference (see

PREFECT_TASKS_REFRESH_CACHE

)