Short answer: The

cache_key

column is only populated when a task is configured to use caching. The env vars you set (

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

and

PREFECT_TASKS_DEFAULT_PERSIST_RESULT=true

) turn on result persistence by default, but they do not enable task caching. With no cache policy set on the task, Prefect won’t compute a cache key, so

task_run.cache_key

remains NULL. What to do if you want cache keys (and caching): - Explicitly set a cache policy on the task via the

@task

decorator. - Optionally set a

cache_expiration

as well. Example:

from datetime import timedelta
import prefect
from prefect import flow, task
from prefect.cache_policies import INPUTS  # or DEFAULT, TASK_SOURCE, etc.

@task(cache_policy=INPUTS, cache_expiration=timedelta(hours=1))
def expensive(x: int) -> int:
    return x * 2

@flow
def my_flow():
    # First run computes and persists result; cache_key gets written
    a = expensive(21)
    # Second run within the same caching window will hit the cache
    b = expensive(21)
    return a, b

if __name__ == "__main__":
    my_flow()

After running with a cache policy, you should see

cache_key

and

cache_expiration

populated for those task runs:

SELECT id, state_id, cache_key, cache_expiration, name
FROM task_run
ORDER BY created DESC
LIMIT 20;

Notes and gotchas: - Caching and result persistence are separate concerns: - Persistence controls whether results are written to storage by default. - Caching controls whether a task can reuse a previous result; this is what produces a

cache_key

. - If you use a policy like

NO_CACHE

, or do not set any

cache_policy

,

cache_key

will be empty. - You can also set a custom

cache_key_fn

if you want full control over the key. Docs for more detail: - Caching overview: Prefect 3 Caching - Results and persistence: Prefect 3 Results If your goal is just to persist results (not to enable caching), your current env vars are sufficient and the empty

cache_key

is expected. If your goal is to enable caching, add a

cache_policy

to your tasks as shown above.

Great goal—and you can get there, but there are two key points to fix: 1) “Only re-run failed tasks” on a user re-run - Use task caching with a cache policy that does NOT include

RUN_ID

. - Persist results to shared storage so subsequent flow runs can see prior results. - Then, when you trigger a new run with the same parameters, tasks that already completed will be skipped (cache hit) and only failed tasks will actually run. A solid policy for your use case is to base the cache on inputs/parameters and task code:

from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import FLOW_PARAMETERS, TASK_SOURCE, INPUTS

@task(
    cache_policy=FLOW_PARAMETERS + TASK_SOURCE + INPUTS,
    cache_expiration=timedelta(hours=3),
    persist_result=True,  # ensure persistence in the runtime process
)
def do_work(x: int) -> int:
    return x * 2

@flow
def my_flow(x: int):
    a = do_work(x)
    return a

- With this, a re-run of

my_flow(x=123)

within 3 hours will use the cached result and skip the task if it already succeeded. If a task failed previously, it won’t have a valid cached result and will re-execute. 2) Why your

cache_key

stays empty and there’s no cache hit - Including

RUN_ID

in your policy (

RUN_ID + FLOW_PARAMETERS

) guarantees different keys for each flow run, so a new run cannot hit the prior run’s cache. This explains “re-running the task does not use the cached result.” - However, the column being empty indicates caching wasn’t actually active for that task run. The most common causes: - The policy object isn’t what you think it is. Make sure you’re importing the Prefect cache policy constants (not strings) and combining them:

from prefect.cache_policies import RUN_ID, FLOW_PARAMETERS
    @task(cache_policy=RUN_ID + FLOW_PARAMETERS, cache_expiration=timedelta(hours=3))

If you wrote

"RUN_ID" + "FLOW_PARAMETERS"

(strings), Prefect won’t enable caching. - Result persistence is not enabled in the runtime process. The env vars you set must be present where the task executes (the worker/flow process), not just on the API server. As a quick test, set

persist_result=True

on the task decorator (as shown above). - The result store is not shared/persistent across runs (e.g., local filesystem in ephemeral containers without a mounted volume). This prevents cross-run cache hits. Configure a shared/remote result store (e.g., S3/GCS/RemoteFileSystem) or mount a persistent volume. - Typos or a different task definition is actually running. What to change right now - Remove

RUN_ID

from your policy; it prevents cross-run reuse:

from prefect.cache_policies import FLOW_PARAMETERS, TASK_SOURCE, INPUTS
  @task(cache_policy=FLOW_PARAMETERS + TASK_SOURCE + INPUTS, cache_expiration=timedelta(hours=3), persist_result=True)

- Ensure the runtime process sees your persistence settings or set

persist_result=True

explicitly. - Use shared result storage across runs. How to verify quickly - Run a minimal flow twice with the same parameters using the revised policy. On the second run: - The task should start in a Cached/Completed path and not re-execute. - Your database

task_run.cache_key

and

task_run.cache_expiration

should be populated for those runs. - You can also query:

SELECT id, name, cache_key, cache_expiration, state_type
    FROM task_run
    ORDER BY created DESC
    LIMIT 20;

Helpful docs - Caching concepts: Prefect 3 Caching - Results/persistence: Prefect 3 Results If you’re still seeing empty

cache_key

after these changes, can you share: - The exact task decorator (with imports) - Prefect version - Where the env vars are set (API server vs. worker/flow container) - Your result storage configuration (local vs. remote/mounted volume) I’ll help you pinpoint the last mile.