validation#

annotated types as a type library#

the most important pattern in pydantic. instead of repeating @field_validator on every model, bind validation to the type itself:

from typing import Annotated
from pydantic import AfterValidator, BeforeValidator, Field

NonNegativeInteger = Annotated[int, Field(ge=0)]
PositiveInteger = Annotated[int, Field(gt=0)]
StatusCode = Annotated[int, Field(ge=100, le=599)]
LogLevel = Annotated[
    Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
    BeforeValidator(lambda x: x.upper()),
]

now any model field typed as StatusCode gets validation for free. the validator travels with the type.

this scales into a full type library:

# types/__init__.py
from .names import Name, VariableName, URILike
from ._datetime import DateTime, PositiveInterval

NonNegativeFloat = Annotated[float, Field(ge=0.0)]
LaxUrl = Annotated[str, BeforeValidator(lambda x: str(x).strip())]

from prefect/types/__init__.py

BeforeValidator vs AfterValidator#

BeforeValidator — runs before pydantic's own parsing. use for coercion (strings to timedeltas, env var parsing):

def _convert_seconds_to_timedelta(value: Any) -> Any:
    if isinstance(value, timedelta):
        return value
    if isinstance(value, (int, float)):
        return timedelta(seconds=value)
    return value

SecondsTimeDelta = Annotated[timedelta, BeforeValidator(_convert_seconds_to_timedelta)]

AfterValidator — runs after pydantic parses. use for constraints on already-typed values:

def _validate_non_negative_timedelta(v: timedelta) -> timedelta:
    if v.total_seconds() < 0:
        raise ValueError("must be non-negative")
    return v

NonNegativeTimedelta = Annotated[timedelta, AfterValidator(_validate_non_negative_timedelta)]

composing validators#

Annotated stacks. you can layer multiple validators and field constraints:

ValidAssetKey = Annotated[
    str,
    AfterValidator(validate_valid_asset_key),
    Field(
        max_length=512,
        description="URI-like string with restricted characters",
    ),
]

domain-specific types in their own module#

when a domain has enough types, extract them. prefect's names.py defines Name, VariableName, BlockDocumentName, URILike — all Annotated types with validators using functools.partial for parameterization:

from functools import partial

def raise_on_name_alphanumeric_dashes_only(
    value: str | None, field_name: str = "value"
) -> str | None:
    if value is not None and not re.match(r"^[a-z0-9-]*$", value):
        raise ValueError(f"{field_name} must only contain lowercase letters, numbers, and dashes.")
    return value

BlockDocumentName = Annotated[
    Name,
    AfterValidator(partial(raise_on_name_alphanumeric_dashes_only, field_name="Block document name")),
]

partial lets one validator function serve multiple domain types with customized error messages.

from prefect/types/names.py

model_validator for side effects#

run setup code when a model is instantiated:

from typing import Self
from pydantic import model_validator
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    debug: bool = False

    @model_validator(mode="after")
    def configure_logging(self) -> Self:
        setup_logging(debug=self.debug)
        return self

settings = Settings()  # logging configured on import

the validator runs after all fields are set. use for side effects that depend on configuration values.

from bot/config.py

env var parsing with BeforeValidator#

for settings that come from environment variables as strings but need richer types:

def validate_set_T_from_delim_string(
    value: str | T | set[T] | None, type_: Any, delim: str | None = None
) -> set[T]:
    """parse comma-delimited env vars into typed sets.

    PREFECT_CLIENT_RETRY_EXTRA_CODES=429,502,503 → {429, 502, 503}
    """
    if not value:
        return set()
    T_adapter = TypeAdapter(type_)
    delim = delim or ","
    if isinstance(value, str):
        return {T_adapter.validate_strings(s.strip()) for s in value.split(delim)}
    ...

ClientRetryExtraCodes = Annotated[
    str | StatusCode | set[StatusCode] | None,
    BeforeValidator(partial(validate_set_T_from_delim_string, type_=StatusCode)),
]

TypeAdapter is useful here — it validates individual values without needing a full model.