<@ULVA73B9P> is it possible to build the flow para...
# ask-marvin
l
@Marvin is it possible to build the flow parameters dynamically using pydantic?
m
thought for 267.4 seconds
Short answer: yes. In Prefect 3.x you can define flow parameters with Pydantic models—including models you build dynamically with
pydantic.create_model
. Prefect uses your function signature to generate an OpenAPI schema for parameters, which the UI uses to render forms, and which deployments use to validate inputs. The key constraint: the schema is fixed by the flow’s signature at definition/import time; you can’t change it per-run. Example: static Pydantic model as a single flow parameter
Copy code
from typing import Literal
from pydantic import BaseModel, Field
from prefect import flow

class IngestParams(BaseModel):
    source: Literal["s3", "gcs", "local"] = "s3"
    path: str = Field(description="Object path or local file path")
    batch_size: int = Field(ge=1, le=10_000, description="Rows per chunk")

@flow
def ingest(params: IngestParams):
    # Inside the flow, `params` is an IngestParams instance (validated & parsed)
    print(params)
Example: dynamically build a parameter model with create_model
Copy code
from typing import Literal
from pydantic import Field, create_model
from prefect import flow

# Imagine these come from env, config, or metadata discovered at import/deploy time
regions = ["us-east-1", "eu-west-1", "ap-southeast-2"]
default_region = "us-east-1"

DynamicParams = create_model(
    "DynamicParams",
    # Required field (ellipsis) and constrained options
    region=(Literal[tuple(regions)], default_region),
    threads=(int, Field(default=4, ge=1, le=64, description="Worker threads")),
    bucket=(str, Field(description="Target bucket")),
)

@flow
def process(params: DynamicParams):
    print(f"Running in {params.region} with {params.threads} threads, bucket={params.bucket}")
You can also attach constraints to primitives with Annotated + Field
Copy code
from typing import Annotated
from pydantic import Field
from prefect import flow

NonEmpty = Annotated[str, Field(min_length=1)]
Port = Annotated[int, Field(ge=1, le=65535)]

@flow
def connect(host: NonEmpty, port: Port = 5432):
    ...
What happens in deployments and the UI - Prefect computes a parameter schema from the flow signature. Nested Pydantic models become nested JSON objects in the schema. - The UI renders forms from that schema (enums from Literal, number inputs with ge/le, etc.) and shows descriptions/defaults. - On run submission, parameters are validated against that schema; at run start, Prefect validates/parses again so your flow receives model instances and constrained primitives. Caveats and tips - Schema is determined once when the flow is defined (import time). If you need “dynamic” fields, build the Pydantic model before defining/decorating the flow (e.g., compute choices from config and then call
create_model
). - Custom Pydantic validators run at flow start; the UI/backend validation uses the generated JSON schema (so some custom logic won’t be enforced until runtime). - Prefect 3 uses Pydantic 2; avoid mixing Pydantic v1 types in the same flow. - You can influence form layout with
Field(..., description="...", json_schema_extra={"position": 0})
to order fields. Docs and references - Prefect forms and parameter schemas: Form building from schemas - Deployments overview: Deployments in Prefect 3 - API reference (schema generation utilities): prefect.utilities.callables If you share a bit about how you want the parameters to be “dynamic” (e.g., vary by environment, loaded from a catalog, driven by a database), I can suggest a concrete pattern for building the model and wiring it into your flow/deployments.