artur
10/24/2025, 1:05 AMMarvin
10/24/2025, 1:05 AMMarvin
10/24/2025, 1:13 AMrefresh_cache is a task-level setting; there’s no flow-level switch to “retry the flow with refresh_cache=True.”
Ways to get the effect on a flow retry:
- Set it on the tasks that should refresh on retry
from prefect import flow, task
@task(retries=3, refresh_cache=True)
def do_work(x):
...
@flow(retries=2)
def my_flow(x):
return do_work(x)
- Turn it on globally for the run/deployment via settings/env var
# e.g., in your worker/deployment environment
export PREFECT_TASKS_REFRESH_CACHE=true
Docs: Settings reference (see tasks.refresh_cache / PREFECT_TASKS_REFRESH_CACHE)
- Make cache keys change on retries so cached results aren’t reused
- Include the flow run attempt in your task’s cache_key_fn to force a new key on each flow retry:
import hashlib, json
from prefect import task
from prefect.runtime import flow_run
def key_with_retry(ctx, task_args):
attempt = flow_run.run_count() or 0
args_hash = hashlib.sha1(json.dumps(task_args, sort_keys=True).encode()).hexdigest()
return f"{ctx.task.name}:{attempt}:{args_hash}"
@task(cache_key_fn=key_with_retry)
def do_work(x):
...
Docs: Caching concepts
- Disable caching on tasks you always want to recompute on retry
from prefect import task
from prefect.cache_policies import NO_CACHE
@task(cache_policy=NO_CACHE, retries=3)
def do_work(x):
...
Notes
- The @flow decorator has retries but no refresh_cache parameter.
- There isn’t a CLI/UI toggle to “retry with refresh_cache” at the flow level.
- Advanced: you can use API state transitions with state_details.refresh_cache=True, but that’s generally only for custom orchestration. Docs: Set flow run state.
If you can share whether you want to recompute everything on retry or only specific steps, I can suggest the minimal change for your flow.