If you’re dealing with too many variables and parameters, the recommended pattern in Prefect 3.x is: don’t keep passing lots of loose arguments around — group them into structured objects and split the flow into logical stages. A good approach is: 1. Group related parameters into config models Use a Pydantic model or dataclass for related settings instead of long function signatures.

from pydantic import BaseModel
from prefect import flow, task

class ExtractConfig(BaseModel):
    api_url: str
    batch_size: int = 100

class TransformConfig(BaseModel):
    strict: bool = False
    normalize_names: bool = True

class PipelineConfig(BaseModel):
    extract: ExtractConfig
    transform: TransformConfig
    run_date: str

@flow
def pipeline(config: PipelineConfig):
    data = extract(config.extract)
    cleaned = transform(data, config.transform)
    load(cleaned)

This is usually the cleanest way to handle “too many params”. 2. Break the flow into subflows or stages If the flow is doing extraction, validation, transformation, and loading, make those separate subflows or clear sections.

@flow
def extract_flow(config: ExtractConfig):
    ...

@flow
def transform_flow(data, config: TransformConfig):
    ...

@flow
def main_flow(config: PipelineConfig):
    data = extract_flow(config.extract)
    result = transform_flow(data, config.transform)
    return result

This gives you: - cleaner code - better observability in the UI - easier testing - fewer giant task signatures 3. Keep tasks small and single-purpose Tasks should usually do one thing: - fetch data - validate a batch - transform a record set - write output Avoid giant “do everything” tasks that require 12 parameters. 4. Pass data objects, not lots of primitive values Instead of:

task(a, b, c, d, e, f, g)

prefer:

task(dataset, config)

where

dataset

and

config

are structured objects. 5. Separate run-time inputs from shared configuration A useful rule of thumb: - Flow parameters: things that change every run - Blocks / Variables: shared config used across runs - Task inputs: actual datasets/results from upstream steps So for example: -

run_date

,

customer_id

→ flow parameters - database credentials, bucket names → Blocks / Variables - extracted records → task outputs passed downstream 6. Use meaningful intermediate models for datasets If you have many related outputs, wrap them in a model instead of returning many separate values.

class ValidationResult(BaseModel):
    valid_records: list[dict]
    invalid_records: list[dict]
    error_count: int

Then downstream tasks can consume

ValidationResult

instead of several parallel variables. 7. Avoid monolithic flows with too much local state If you have dozens of local variables in one flow body, that’s usually a sign you should: - create subflows - extract helper functions - define result/config models - simplify task boundaries Practical recommendation A solid structure is: - one top-level

PipelineConfig

- nested config models by concern - subflows for major pipeline phases - tasks that take

config

+ one main input dataset - structured result objects for complex outputs So instead of this:

@flow
def my_flow(
    api_url, batch_size, strict, normalize_names,
    output_table, retries, timeout, run_date, region, ...
):
    ...

prefer this:

class PipelineConfig(BaseModel):
    extract: ExtractConfig
    transform: TransformConfig
    load: LoadConfig
    run_date: str

@flow
def my_flow(config: PipelineConfig):
    ...

That’s the pattern I’d recommend most strongly. Relevant docs: - Flows - Tasks - Blocks - Settings and profiles If you want, I can also show a before/after refactor example of a messy flow into a cleaner Prefect structure.