Lester Cayabyab
10/17/2025, 3:34 PMMarvin
10/17/2025, 3:34 PMMarvin
10/17/2025, 3:38 PMpydantic.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
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
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
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.