about things
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

at main 156 lines 5.0 kB view raw view rendered
1# validation 2 3## annotated types as a type library 4 5the most important pattern in pydantic. instead of repeating `@field_validator` on every model, bind validation to the type itself: 6 7```python 8from typing import Annotated 9from pydantic import AfterValidator, BeforeValidator, Field 10 11NonNegativeInteger = Annotated[int, Field(ge=0)] 12PositiveInteger = Annotated[int, Field(gt=0)] 13StatusCode = Annotated[int, Field(ge=100, le=599)] 14LogLevel = Annotated[ 15 Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], 16 BeforeValidator(lambda x: x.upper()), 17] 18``` 19 20now any model field typed as `StatusCode` gets validation for free. the validator travels with the type. 21 22this scales into a full type library: 23 24```python 25# types/__init__.py 26from .names import Name, VariableName, URILike 27from ._datetime import DateTime, PositiveInterval 28 29NonNegativeFloat = Annotated[float, Field(ge=0.0)] 30LaxUrl = Annotated[str, BeforeValidator(lambda x: str(x).strip())] 31``` 32 33from [prefect/types/\_\_init\_\_.py](https://github.com/prefecthq/prefect/blob/main/src/prefect/types/__init__.py) 34 35## BeforeValidator vs AfterValidator 36 37- `BeforeValidator` — runs before pydantic's own parsing. use for coercion (strings to timedeltas, env var parsing): 38 39```python 40def _convert_seconds_to_timedelta(value: Any) -> Any: 41 if isinstance(value, timedelta): 42 return value 43 if isinstance(value, (int, float)): 44 return timedelta(seconds=value) 45 return value 46 47SecondsTimeDelta = Annotated[timedelta, BeforeValidator(_convert_seconds_to_timedelta)] 48``` 49 50- `AfterValidator` — runs after pydantic parses. use for constraints on already-typed values: 51 52```python 53def _validate_non_negative_timedelta(v: timedelta) -> timedelta: 54 if v.total_seconds() < 0: 55 raise ValueError("must be non-negative") 56 return v 57 58NonNegativeTimedelta = Annotated[timedelta, AfterValidator(_validate_non_negative_timedelta)] 59``` 60 61## composing validators 62 63`Annotated` stacks. you can layer multiple validators and field constraints: 64 65```python 66ValidAssetKey = Annotated[ 67 str, 68 AfterValidator(validate_valid_asset_key), 69 Field( 70 max_length=512, 71 description="URI-like string with restricted characters", 72 ), 73] 74``` 75 76## domain-specific types in their own module 77 78when 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: 79 80```python 81from functools import partial 82 83def raise_on_name_alphanumeric_dashes_only( 84 value: str | None, field_name: str = "value" 85) -> str | None: 86 if value is not None and not re.match(r"^[a-z0-9-]*$", value): 87 raise ValueError(f"{field_name} must only contain lowercase letters, numbers, and dashes.") 88 return value 89 90BlockDocumentName = Annotated[ 91 Name, 92 AfterValidator(partial(raise_on_name_alphanumeric_dashes_only, field_name="Block document name")), 93] 94``` 95 96`partial` lets one validator function serve multiple domain types with customized error messages. 97 98from [prefect/types/names.py](https://github.com/prefecthq/prefect/blob/main/src/prefect/types/names.py) 99 100## model_validator for side effects 101 102run setup code when a model is instantiated: 103 104```python 105from typing import Self 106from pydantic import model_validator 107from pydantic_settings import BaseSettings 108 109class Settings(BaseSettings): 110 debug: bool = False 111 112 @model_validator(mode="after") 113 def configure_logging(self) -> Self: 114 setup_logging(debug=self.debug) 115 return self 116 117settings = Settings() # logging configured on import 118``` 119 120the validator runs after all fields are set. use for side effects that depend on configuration values. 121 122from [bot/config.py](https://github.com/zzstoatzz/bot) 123 124## env var parsing with BeforeValidator 125 126for settings that come from environment variables as strings but need richer types: 127 128```python 129def validate_set_T_from_delim_string( 130 value: str | T | set[T] | None, type_: Any, delim: str | None = None 131) -> set[T]: 132 """parse comma-delimited env vars into typed sets. 133 134 PREFECT_CLIENT_RETRY_EXTRA_CODES=429,502,503 → {429, 502, 503} 135 """ 136 if not value: 137 return set() 138 T_adapter = TypeAdapter(type_) 139 delim = delim or "," 140 if isinstance(value, str): 141 return {T_adapter.validate_strings(s.strip()) for s in value.split(delim)} 142 ... 143 144ClientRetryExtraCodes = Annotated[ 145 str | StatusCode | set[StatusCode] | None, 146 BeforeValidator(partial(validate_set_T_from_delim_string, type_=StatusCode)), 147] 148``` 149 150`TypeAdapter` is useful here — it validates individual values without needing a full model. 151 152from [prefect/types/\_\_init\_\_.py](https://github.com/prefecthq/prefect/blob/main/src/prefect/types/__init__.py) 153 154sources: 155- [coping with python's type system](https://blog.zzstoatzz.io/coping-with-python-type-system/) 156- [prefect/types/](https://github.com/prefecthq/prefect/tree/main/src/prefect/types)