Short answer: In Prefect 3.x you can cache tasks that take Pydantic models as arguments without special handling—Prefect’s cache key hashing knows how to serialize Pydantic models deterministically. Just pick the right cache policy. Key points - By default, Prefect’s cache key includes: task inputs, task source code, and the current flow run ID. - If you want cache reuse across different flow runs, use the Inputs-only policy. - If your model contains volatile fields (timestamps, request IDs), use a custom cache key function to exclude them. Examples 1) Basic caching with a Pydantic model (shared across flow runs)

from datetime import datetime
from pydantic import BaseModel
from prefect import flow, task
from prefect.cache_policies import INPUTS  # Inputs-only: reuses cache across runs

class UserConfig(BaseModel):
    user_id: int
    region: str
    requested_at: datetime | None = None

@task(cache_policy=INPUTS, persist_result=True)
def fetch_user(cfg: UserConfig) -> dict:
    print("executing...")
    # Simulate expensive work here
    return {"greeting": f"hello {cfg.user_id} in {cfg.region}"}

@flow
def main():
    # First call computes, subsequent identical cfg values hit cache (even in future runs)
    a = fetch_user(UserConfig(user_id=1, region="us"))
    b = fetch_user(UserConfig(user_id=1, region="us"))
    return a, b

if __name__ == "__main__":
    main()

2) Default policy (cache is scoped to the current flow run)

from prefect.cache_policies import DEFAULT  # = Inputs + TaskSource + RunId

@task(cache_policy=DEFAULT, persist_result=True)
def compute(cfg: UserConfig) -> str:
    print("executing...")
    return f"{cfg.user_id}:{cfg.region}"

Note: With DEFAULT, identical calls in the same run hit cache; a new flow run won’t reuse the cache. 3) Exclude volatile top-level parameters If your volatile value is a separate parameter (not inside the model), you can exclude it from the Inputs policy:

from prefect.cache_policies import Inputs

@task(cache_policy=Inputs() - "timestamp", persist_result=True)
def load_things(cfg: UserConfig, timestamp: str) -> list[str]:
    print("executing...")
    return ["a", "b"]

4) Exclude volatile fields inside the Pydantic model (custom cache key) When the volatile field lives inside the model, use a custom cache-key function:

from prefect.cache_policies import CacheKeyFnPolicy
from prefect.utilities.hashing import hash_objects

def cfg_cache_key(_, params: dict) -> str:
    cfg = params.get("cfg")
    # Make a stable dict and exclude volatile fields
    if isinstance(cfg, UserConfig):
        stable = cfg.model_dump(mode="json", exclude={"requested_at"})
    else:
        stable = cfg
    return hash_objects({"cfg": stable}, raise_on_failure=True)

@task(cache_policy=CacheKeyFnPolicy(cache_key_fn=cfg_cache_key), persist_result=True)
def fetch_user(cfg: UserConfig) -> dict:
    print("executing...")
    return {"greeting": f"hello {cfg.user_id} in {cfg.region}"}

Notes and tips - Prefect 3.x Pydantic support: Prefect’s hashing uses JSON serialization for BaseModel (v2) which yields consistent cache keys for identical content. If JSON fails, it falls back to cloudpickle. - Scope: Choose

INPUTS

if you want cache reuse across different runs; choose

DEFAULT

if you want cache per-run by design. - Results:

persist_result=True

ensures the result is stored so future calls can retrieve it (you can also set

result_storage

and

result_serializer

if you need a specific backend). - If you need to ignore multiple nested fields, just expand the

exclude

set in

model_dump

. If you share a quick snippet of your model and how you’d like the cache to behave (per-run vs cross-run, fields to ignore), I can tailor the exact cache policy or key function for you.