Great question. In Prefect 3.x, you can do this by providing a custom

cache_key_fn

on your task. Return a string to enable caching for that run, or return

None

to skip caching. So to “activate cache only when a parameter is None”, have the function return a key only in that case. Here’s a minimal example:

from datetime import timedelta
from prefect import flow, task
from prefect.context import TaskRunContext

def cache_when_none(ctx: TaskRunContext, params: dict):
    # Only cache when `user_id` is None
    if params.get("user_id") is None:
        # Use a stable key; could also incorporate other params if needed
        return "user_id=None"
    return None  # Bypass caching for all other values

@task(cache_key_fn=cache_when_none, cache_expiration=timedelta(hours=6))
def get_profile(user_id: str | None):
    print("Computing...")
    return {"user_id": user_id, "data": 42}

@flow
def main():
    a = get_profile(None)        # caches
    b = get_profile(None)        # cache hit
    c = get_profile("alice")     # bypasses cache
    d = get_profile(None, refresh_cache=True)  # bypass cache this time

Notes: - Returning

None

from

cache_key_fn

disables caching for that run. -

cache_expiration

is optional; include it if you want time-based invalidation. -

persist_result

defaults to auto-enable when a cache function/policy is set; if you set

persist_result=False

, caching will be disabled entirely. - You can force recomputation on a single call with

refresh_cache=True

. If you want the cache key to reflect other parameters when

user_id

is None, you can derive it from them:

from prefect.utilities.hashing import hash_objects

def cache_when_none(ctx: TaskRunContext, params: dict):
    if params.get("user_id") is None:
        other = {k: v for k, v in params.items() if k != "user_id"}
        return f"none:{hash_objects(other)}"
    return None

Want to target a different parameter name or more complex logic? Share your task signature and I can tailor the

cache_key_fn

for you.

Great instinct. If you need to “play nice” with other policies and still enforce “only cache when param is None,” the safest approach is to write a custom CachePolicy that wraps the other policies and gates them behind your condition. Why: in Prefect 3.x, CompoundCachePolicy (the thing behind

INPUTS + TASK_SOURCE + ...

) ignores None-returning components and hashes the rest. That means a component cannot veto caching. Caching is only disabled if all components return None. So if you compose your conditional policy with others using

+

, those others can still produce a key and you’ll cache even when your condition says not to. Recommended pattern: create a wrapper policy that returns None when the condition is not met, and otherwise delegates to a composed “inner” policy. Example: only cache when

user_id

is None, and when caching, use INPUTS + TASK_SOURCE.

from dataclasses import dataclass
from typing import Any, Optional
from prefect import task, flow
from prefect.cache_policies import CachePolicy, INPUTS, TASK_SOURCE
from prefect.context import TaskRunContext

@dataclass
class CacheOnlyWhenNone(CachePolicy):
    param: str = "user_id"
    inner: CachePolicy = INPUTS + TASK_SOURCE  # policies to use when condition is met

    def compute_key(
        self,
        ctx: TaskRunContext,
        inputs: dict[str, Any],
        flow_parameters: dict[str, Any],
        **_,
    ) -> Optional[str]:
        # If the parameter is NOT None, disable caching entirely
        if inputs.get(self.param) is not None:
            return None
        # Otherwise, delegate to the inner policy
        return self.inner.compute_key(ctx, inputs, flow_parameters)

@task(cache_policy=CacheOnlyWhenNone(param="user_id", inner=INPUTS + TASK_SOURCE))
def get_profile(user_id: str | None):
    print("Computing...")
    return {"user_id": user_id, "data": 42}

@flow
def main():
    a = get_profile(None)        # caches
    b = get_profile(None)        # cache hit
    c = get_profile("alice")     # no cache

Notes: - Avoid composing your conditional policy with

+

outside the wrapper if you need veto semantics; put the other policies inside the wrapper via the

inner

field. - If you don’t need composition, a

cache_key_fn

is simpler. But if you compose

CacheKeyFnPolicy + OTHER

, your function cannot veto caching—OTHER can still produce a key. - You can still configure storage/isolation on your custom policy with

.configure(...)

if needed. Docs: - Caching guide: Cache workflow steps - Cache policy API: CachePolicy API If you share which other policies you want included (e.g., RUN_ID, FLOW_PARAMETERS), I can tailor the wrapper to match your setup.

You should pass an initialized CachePolicy object to the task decorator, not the class. The

@task

signature expects a CachePolicy instance. Passing the class itself will not work. Here’s a minimal, working example of the “gate + delegate” pattern that vetoes caching unless a parameter is None, while still composing with other policies when caching is allowed:

from dataclasses import dataclass
from typing import Any, Optional
from datetime import timedelta

from prefect import task, flow
from prefect.cache_policies import CachePolicy, INPUTS, TASK_SOURCE
from prefect.context import TaskRunContext

@dataclass
class CacheOnlyWhenNone(CachePolicy):
    param: str = "user_id"
    inner: CachePolicy = INPUTS + TASK_SOURCE  # policies used when condition is met

    def compute_key(
        self,
        ctx: TaskRunContext,
        inputs: dict[str, Any],
        flow_parameters: dict[str, Any],
        **_,
    ) -> Optional[str]:
        # If the parameter is NOT None, disable caching entirely
        if inputs.get(self.param) is not None:
            return None
        # Otherwise, delegate to the inner policy
        return self.inner.compute_key(ctx, inputs, flow_parameters)

# IMPORTANT: pass an INSTANCE, not the class
policy = CacheOnlyWhenNone(param="user_id", inner=INPUTS + TASK_SOURCE)

@task(cache_policy=policy, cache_expiration=timedelta(hours=6))
def get_profile(user_id: str | None):
    print("Computing...")
    return {"user_id": user_id, "data": 42}

@flow
def main():
    a = get_profile(None)        # caches (prints "Computing..." once)
    b = get_profile(None)        # cache hit (no print)
    c = get_profile("alice")     # bypass cache (prints "Computing...")

Checklist if it “doesn’t work”: - Ensure you pass

cache_policy=CacheOnlyWhenNone(...)

(an instance), not

cache_policy=CacheOnlyWhenNone

. - Don’t set

persist_result=False

anywhere; that disables caching entirely. - If you also set

cache_key_fn

, it will override `cache_policy`—remove it for this pattern. - For a single-call bypass, use

refresh_cache=True

. Docs: - Caching guide: Cache workflow steps - CachePolicy API: CachePolicy API If you still see issues, can you share your exact decorator line and any logs/tracebacks? I’ll pinpoint what’s going on.

You’re hitting that TypeError because Prefect calls CachePolicy.compute_key with keyword arguments named exactly task_ctx, inputs, and flow_parameters. Your method must either: - use the exact name task_ctx, or - make your first param optional (e.g., ctx=None) and read task_ctx from **kwargs. Here’s a fixed version that matches the engine’s call signature and delegates correctly:

from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import timedelta

from prefect import task, flow
from prefect.cache_policies import CachePolicy, INPUTS, TASK_SOURCE
from prefect.context import TaskRunContext

@dataclass
class CacheOnlyWhenNone(CachePolicy):
    param: str = "user_id"
    inner: CachePolicy = field(default_factory=lambda: INPUTS + TASK_SOURCE)

    def compute_key(
        self,
        task_ctx: TaskRunContext,
        inputs: dict[str, Any],
        flow_parameters: dict[str, Any],
        **kwargs: Any,
    ) -> Optional[str]:
        # veto caching if the parameter is not None
        if inputs.get(self.param) is not None:
            return None
        # otherwise, delegate to the composed policy
        return self.inner.compute_key(
            task_ctx=task_ctx,
            inputs=inputs,
            flow_parameters=flow_parameters,
        )

policy = CacheOnlyWhenNone(param="user_id")

@task(cache_policy=policy, cache_expiration=timedelta(hours=6))
def get_profile(user_id: str | None):
    print("Computing...")
    return {"user_id": user_id, "data": 42}

@flow
def main():
    a = get_profile(None)        # caches
    b = get_profile(None)        # cache hit
    c = get_profile("alice")     # bypass cache

Quick checks: - Pass an instance to

@task(cache_policy=...)

, not the class. - Don’t also set

cache_key_fn

(it overrides

cache_policy

). - Make sure

persist_result

is not False anywhere (that disables caching). Reference: CachePolicy API If you still see errors, please share your policy class and the exact

@task(...)

line and I’ll pinpoint the mismatch.