Great question. Short answer: in Prefect 3.x the community best practice is to avoid shuttling a big, mutable settings object between tasks. Instead: - Store rich config as a custom Block (Pydantic model) and load it where needed - Treat per-run “edits” as dataflow: return a new model from a task and pass it to downstream tasks - Only persist edits back to a Block if you intend to change the long‑lived configuration; otherwise keep updates scoped to the run If you do pass a Pydantic model across machines, lean on Prefect’s result persistence + serializers. Recommended patterns 1) Pass-and-return (per-run edits; no global side-effects) - Use a Pydantic model as a normal task argument/return value - Ensure results can be read across machines by enabling persistence and choosing a serializer - Prefer returning a new instance vs mutating in-place to avoid confusion with retries/caching Example:

from prefect import flow, task
from pydantic import BaseModel, Field

class Settings(BaseModel):
    region: str
    flags: dict = Field(default_factory=dict)

@task(persist_result=True, result_serializer="json")
def tweak(settings: Settings) -> Settings:
    # Prefer producing a new object
    new_flags = {**settings.flags, "use_fast_path": True}
    return Settings(region=settings.region, flags=new_flags)

@task(persist_result=True, result_serializer="json")
def use(settings: Settings) -> str:
    return f"Running in {settings.region} with {settings.flags}"

@flow(result_serializer="json")
def run():
    s = Settings(region="us-east-1")
    s2 = tweak(s)          # serialized + persisted between machines
    out = use(s2)
    return out

Notes: - JSON works well if your model’s class is importable on all workers; pickle is most flexible but opaque. For large payloads, consider

"compressed/pickle"

. - Ensure your

Settings

class is importable by all workers (i.e., in a module on PYTHONPATH), especially if you use JSON which relies on import paths. 2) Blocks for long‑lived config (single source of truth) - Define a custom Block (Pydantic model) for configuration - Load in each task; this avoids passing config around and ensures cross-machine consistency - If you truly want to persist a configuration change, modify and save the Block explicitly Example:

from prefect import flow, task
from prefect.blocks.core import Block
from pydantic import SecretStr

class AppConfig(Block):
    env: str
    feature_flags: dict
    api_key: SecretStr

# One-time setup (outside of the flow)
# AppConfig(env="prod", feature_flags={}, api_key=SecretStr("...")).save("prod", overwrite=True)

@task
def read_config(name: str) -> AppConfig:
    return AppConfig.load(name)

@task
def maybe_update_config(cfg: AppConfig, flag: str, value: bool, persist: bool = False) -> AppConfig:
    updated = cfg.model_copy(update={"feature_flags": {**cfg.feature_flags, flag: value}})
    if persist:
        updated.save("prod", overwrite=True)  # Explicit side-effect
    return updated

@flow
def run():
    cfg = read_config("prod")
    cfg2 = maybe_update_config(cfg, "use_fast_path", True, persist=False)  # per-run change
    ...

Cautions: - Treat Blocks as configuration, not as a highly mutable “state store” during runs. If many runs may update shared config, consider concurrency/locking or use a dedicated DB/service for shared state. - Use `SecretStr`/secret fields for sensitive config (Prefect redacts them). When to choose which - Pass-and-return: you want per-run edits (no shared side-effects), clear lineage, reproducibility, and safety with retries/caching. - Blocks: you want a single source of truth for shared, long‑lived config that runs load consistently. Perform explicit writes only when you mean to change the shared config. - Variables: for simple, frequently-updated scalar/JSON values (feature flags, timeouts) without defining a Block. Serializer and storage tips

- Set

persist_result=True

on tasks that exchange data across machines, or configure it globally. - Choose a serializer: - `"pickle"`: most flexible for arbitrary Python objects - `"json"`: human-readable, portable; requires model classes importable everywhere -

"compressed/pickle"

or

"compressed/json"

for large payloads - Choose a result store (e.g., S3/GCS) accessible to all workers (

result_storage

at task/flow level or via default config). Common pitfalls - In-place mutation: prefer returning a new model to avoid surprises with retries/caching. - Importability: JSON deserialization requires the model’s import path to be resolvable on all workers. - Shared config updates: avoid implicit side-effects; if needed, write back to a Block explicitly and consider race conditions. Docs and references - Results and serializers: Prefect 3 - Results and serializers - Blocks (custom types): Custom Blocks - Variables: Variables guide - Serializers API: Serializers API - Blocks API: Blocks API Quick recommendation - For most teams: define config as a Block, load it where needed; for per-run tweaks, return a new model and pass it downstream; only write back to a Block when you truly want to change shared config. Would you like me to sketch your specific model as a Block and wire it into your flow, or are you leaning toward the pass-and-return pattern? Also, what result store (S3/GCS/local) do your workers share?

Got it — you need to modify the config based on inputs/parameters. The most common community pattern in Prefect 3.x is: - Keep a stable, long‑lived “base” config in a Block - Build a per‑run “effective” config early in the flow by merging: - the base Block - deployment/flow parameters - runtime context (e.g., flow run tags, job variables) - Pass that effective config to tasks (returning new copies when you tweak it) - Only persist back to a Block if you intentionally want to update shared config (ideally in one, final step) This gives you deterministic runs, avoids race conditions, and still lets you compute config dynamically. Reference implementation

from prefect import flow, task
from prefect.blocks.core import Block
from pydantic import BaseModel, Field, SecretStr
from typing import Optional
import os
from prefect import runtime

# 1) Long-lived base config as a Block
class BaseConfig(Block):
    env: str
    endpoint: str
    default_flags: dict = Field(default_factory=dict)
    api_key: Optional[SecretStr] = None

# one-time:
# BaseConfig(env="prod", endpoint="<https://api.example.com>", default_flags={"retries": 3}).save("base", overwrite=True)

# 2) Per-run effective config (not a Block)
class EffectiveConfig(BaseModel):
    env: str
    endpoint: str
    flags: dict = Field(default_factory=dict)
    api_key: Optional[str] = None     # materialized secret if needed
    region: Optional[str] = None      # example: derived from params

def derive_effective_config(base: BaseConfig, params: dict) -> EffectiveConfig:
    # Example of runtime-aware derivation
    # - read flow params
    # - consider tags/job variables
    # - consider env vars
    tags = set(runtime.flow_run.tags)
    job_vars = runtime.flow_run.job_variables or {}

    region = params.get("region") or job_vars.get("REGION") or os.getenv("REGION") or "us-east-1"
    flags = {**base.default_flags, "fast_path": params.get("fast_path", False)}
    if "canary" in tags:
        flags["retries"] = 0

    return EffectiveConfig(
        env=base.env,
        endpoint=base.endpoint,
        flags=flags,
        api_key=base.api_key.get_secret_value() if base.api_key else None,
        region=region,
    )

@task(persist_result=True, result_serializer="json")
def build_effective_config(base_block_name: str, params: dict) -> EffectiveConfig:
    base = BaseConfig.load(base_block_name)
    return derive_effective_config(base, params)

# 3) Tasks consume and may tweak the config by returning a new copy
@task(persist_result=True, result_serializer="json")
def enrich_config(cfg: EffectiveConfig) -> EffectiveConfig:
    # Prefer producing a new object vs in-place mutation
    new_flags = {**cfg.flags, "use_cache": True}
    return cfg.model_copy(update={"flags": new_flags})

@task
def do_work(cfg: EffectiveConfig):
    # Use fields directly
    ...

# 4) Optional: persist back specific changes to a Block (explicit side-effect)
@task
def maybe_persist_back(cfg: EffectiveConfig, persist: bool = False):
    if not persist:
        return
    base = BaseConfig.load("base")
    # Whitelist only the pieces you actually want to promote to shared config
    updated = base.model_copy(update={"default_flags": cfg.flags})
    updated.save("base", overwrite=True)

@flow(result_serializer="json")
def pipeline(region: str = "us-east-1", fast_path: bool = False, persist_changes: bool = False):
    # Pass any params that influence config
    params = {"region": region, "fast_path": fast_path}
    cfg = build_effective_config("base", params)
    cfg = enrich_config(cfg)
    do_work(cfg)
    maybe_persist_back(cfg, persist=persist_changes)

Why this pattern works - Single source of truth: Base Block remains stable and auditable. - Deterministic runs: Effective config is computed from explicit inputs and runtime context. - Safe mutation: Tasks return a new config instance; no in‑place mutation surprises with retries/caching.

- Cross-machine ready:

persist_result=True

+ a serializer (JSON or pickle) ensures configs move between workers. If multiple tasks need to “contribute” changes - Use a “patch + reducer” approach: - Let tasks return small “patches” (dicts) describing their intended changes. - Merge them deterministically in one dedicated task (avoids races and hidden ordering bugs). - If you truly need concurrent, cross-task shared state, use a transactional store (DB/Redis) rather than updating a Block repeatedly during the run. Serializer/storage guidance - Set

persist_result=True

on the config-producing/consuming tasks. - Use

result_serializer="json"

if your models are importable on all workers;

"pickle"

for maximal flexibility;

"compressed/*"

for large payloads. - Configure a shared result store (S3/GCS/etc.) if tasks run on different machines. Docs - Results and serializers: Results and serializers - Custom Blocks: Custom Blocks - Variables (if you want a lightweight overrides layer): Variables If you share your actual model/inputs, I can sketch a concrete “builder + patches” version that mirrors your needs.

Short answer: No — you don’t need Redis just to scale Prefect. What you need to scale across machines - A worker fleet (K8s/ECS/etc.) and a shared result store (S3/GCS/Azure/local FS) so task results can be persisted/loaded across machines. - Prefect’s result persistence and serializers handle passing your Pydantic config or any data between tasks. When Redis (or another external store) helps - Only if you need many concurrent tasks to update the same shared, mutable state mid-run with coordination/atomicity (e.g., counters, locks, dedupe sets, dynamic work queues). - In those cases, use Redis/Postgres/etc. as an intentional state service. Blocks are best as long‑lived config, not a frequently-updated run-time state. Best-practice for your “modify config based on inputs” need - Keep a stable base config in a Block. - Build an “effective” per-run config early from base + parameters + runtime context. - Pass that to tasks and return new copies when modifying. - Optionally persist back to the Block once at the end (explicit side-effect). No Redis needed. Patch + reducer pattern (if multiple tasks contribute changes)

from prefect import flow, task
from pydantic import BaseModel, Field

class Effective(BaseModel):
    flags: dict = Field(default_factory=dict)

@task
def propose_patch_a(cfg: Effective) -> dict:
    return {"use_cache": True}

@task
def propose_patch_b(cfg: Effective) -> dict:
    return {"retries": 5}

@task
def reduce_patches(cfg: Effective, patches: list[dict]) -> Effective:
    merged = {**cfg.flags}
    for p in patches:
        merged.update(p)
    return cfg.model_copy(update={"flags": merged})

@flow
def run():
    cfg = Effective(flags={"retries": 3})
    p1 = propose_patch_a(cfg)
    p2 = propose_patch_b(cfg)
    cfg2 = reduce_patches(cfg, [p1, p2])  # deterministic merge, no shared mutable state
    return cfg2

Links - Results and serializers: Results and serializers - Custom Blocks: Custom Blocks - Variables: Variables If you describe the exact “shared edits” you need (who updates what, how often, and whether order matters), I can recommend either a pure Prefect pattern (patch + reducer) or when an external store like Redis is justified.