12 Must‑Try Pydantic v2 Model Patterns for Safer Python Code
This guide presents twelve practical Pydantic v2 patterns—from a base DTO and snake/camel case handling to computed fields, immutable objects, configuration management, and endpoint protection—showing concrete code examples and a complete data‑ingestion pipeline that demonstrates how each pattern improves validation, serialization, and overall robustness in Python projects.
Pydantic v2 adds industrial‑grade validation, serialization, and structuring while preserving its original simplicity. The author shares twelve reusable patterns that have been applied in real projects, covering everything from a base DTO to immutable value objects and endpoint protection.
1. Base DTO Model: Friendly Collaboration Across Components
Create a base model with sensible defaults so every model behaves predictably during serialization.
from typing import Any
from pydantic import BaseModel, ConfigDict
class DTO(BaseModel):
model_config = ConfigDict(
from_attributes=True, # Convert ORM objects to models
populate_by_name=True, # Support snake_case and aliases
extra='forbid', # Fail fast on unknown fields
str_strip_whitespace=True # Auto‑strip whitespace from strings
)
class UserDTO(DTO):
id: int
email: str
full_name: str | None = NoneThis keeps behavior consistent across hundreds of models and reduces unexpected serialization results.
2. Easy Snake‑Case ↔ Camel‑Case Handling
APIs often prefer camelCase while Python prefers snake_case. This pattern builds a bridge.
import re
from pydantic import BaseModel, ConfigDict, Field
def to_camel(s: str) -> str:
return re.sub(r'_([a-z])', lambda m: m.group(1).upper(), s)
class ApiModel(BaseModel):
model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
class Product(ApiModel):
product_id: int = Field(alias='productId') # Explicit alias when needed
unit_price: float
# Example usage
Product.model_validate({'productId': 7, 'unitPrice': 3.5}) # ✅3. Field Validators for Trivial Issues
v2 introduces @field_validator with mode='before'|'after' to clean whitespace, enforce case, and catch simple parsing bugs.
from pydantic import BaseModel, field_validator
class Email(BaseModel):
address: str
@field_validator('address', mode='before')
@classmethod
def normalize(cls, v: str) -> str:
v = v.strip().lower()
if '@' not in v:
raise ValueError('Invalid email')
return vAbout 90 % of bugs stem from whitespace, case, or simple parsing problems; this solves them.
4. Cross‑Field Model Validation
When one field depends on another, use @model_validator.
from pydantic import BaseModel, model_validator
class Window(BaseModel):
start: int
end: int
@model_validator(mode='after')
def check_order(self):
if self.end <= self.start:
raise ValueError('end must be > start')
return selfThis keeps business rules close to the data instead of scattering them across services.
5. Computed Fields for Derived Values
Derive values without storing them.
from pydantic import BaseModel, computed_field
class Name(BaseModel):
first: str
last: str
@computed_field
@property
def display(self) -> str:
return f"{self.first.title()}{self.last.title()}"Ideal for UI payloads where derived data should be included automatically.
6. Fine‑Tuned Serialization with Field Serializers
Customize output while preserving rich internal types.
from datetime import datetime, timezone
from pydantic import BaseModel, field_serializer
class Event(BaseModel):
id: str
at: datetime
@field_serializer('at')
def iso8601(self, dt: datetime, _info):
return dt.astimezone(timezone.utc).isoformat()
e = Event(id='a1', at=datetime.now())
print(e.model_dump(by_alias=True, exclude_none=True)) # Custom output
print(e.model_dump_json()) # Quick JSON serialization7. Discriminated Union for Self‑Describing APIs
Model "one‑of‑many" payloads safely without fragile if/else trees.
from typing import Annotated, Union, Literal
from pydantic import BaseModel, Field
class Click(BaseModel):
kind: Literal['click']
x: int
y: int
class Input(BaseModel):
kind: Literal['input']
value: str
Event = Annotated[Union[Click, Input], Field(discriminator='kind')]
def handle(evt: Event):
if evt.kind == 'click':
print(f"Clicked at ({evt.x}, {evt.y})")
else:
print(f"Input: {evt.value}")8. TypeAdapter for Ad‑Hoc Validation
Validate arbitrary types (lists, primitives, nested dicts) on the fly.
from typing import List
from pydantic import TypeAdapter
ta = TypeAdapter(List[int])
nums = ta.validate_python(['1', 2, 3]) # → [1, 2, 3]
json_ready = ta.dump_python(nums) # Fast serializer9. Settings Management with pydantic‑settings
Load configuration from environment variables and .env files, following the Twelve‑Factor app pattern.
from pydantic_settings import BaseSettings, SettingsConfigDict
class AppSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='APP_', env_file='.env')
db_url: str
debug: bool = False
cache_ttl: int = 300
# Usage: APP_DB_URL=postgres://... python app.py
settings = AppSettings()10. ORM Interoperability without Boilerplate
v2 replaces from_orm=True with from_attributes=True and model_validate for clean DTO creation from ORM rows.
from pydantic import BaseModel, ConfigDict
class User(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
email: str
# Example: user_row = SomeORM.query.first()
# u = User.model_validate(user_row)11. Trusted Immutable Value Objects
Some data should never change after creation (e.g., IDs, amounts, coordinates).
from pydantic import BaseModel, ConfigDict
class Money(BaseModel):
model_config = ConfigDict(frozen=True)
amount: int # cents
currency: str = 'USD'
m = Money(amount=500)
# m.amount = 600 # ❌ raises error
m2 = m.model_copy(update={'amount': 600}) # ✅ new instance12. Validate Call for Boundary Protection
Guard service functions, CLI entry points, or scheduled jobs with @validate_call.
from pydantic import validate_call, Field
from typing import Annotated
@validate_call
def charge(user_id: int, amount: Annotated[float, Field(gt=0)]):
return {'ok': True}
charge(42, '9.99') # ✅ converts to 9.99
# charge('u1', -5) # ❌ raises ValidationErrorComplete Data‑Ingestion Pipeline
The following script combines all twelve patterns into a production‑ready CSV‑to‑JSON pipeline.
import csv, json
from pathlib import Path
from datetime import datetime
from typing import List, Annotated
from pydantic import (
BaseModel, ConfigDict, Field, field_validator, model_validator,
TypeAdapter, validate_call
)
from pydantic_settings import BaseSettings, SettingsConfigDict
# 9. Settings management
class AppSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='INGEST_', env_file='.env')
input_path: str = "data/input.csv"
output_path: str = "data/output.json"
strict_validation: bool = True
batch_size: int = 1000
# 11. Immutable identifier
class RecordId(BaseModel):
model_config = ConfigDict(frozen=True)
value: str
@field_validator('value')
@classmethod
def validate_id(cls, v: str) -> str:
if not v or len(v) > 50:
raise ValueError('ID must be non‑empty and under 50 chars')
return v.strip()
# 2. Naming conversion helper
def to_camel(s: str) -> str:
parts = s.split('_')
return parts[0] + ''.join(p.title() for p in parts[1:])
class ApiModel(BaseModel):
model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
# 3‑4‑5‑6‑7‑8‑12 combined model
class CustomerRecord(ApiModel):
record_id: RecordId
first_name: str
last_name: str
email: str
age: int
signup_date: datetime
vip_status: bool = False
@field_validator('email', mode='before')
@classmethod
def normalize_email(cls, v: str) -> str:
v = v.strip().lower()
if '@' not in v:
raise ValueError('Invalid email format')
return v
@field_validator('age')
@classmethod
def validate_age(cls, v: int) -> int:
if v < 0 or v > 150:
raise ValueError('Age must be between 0 and 150')
return v
@model_validator(mode='after')
def check_business_rules(self):
if self.vip_status and self.age < 18:
raise ValueError('VIP customers must be at least 18 years old')
return self
@computed_field
@property
def full_name(self) -> str:
return f"{self.first_name}{self.last_name}"
@computed_field
@property
def customer_segment(self) -> str:
if self.age < 25:
return "young"
elif self.age < 45:
return "adult"
else:
return "senior"
class OutputPayload(ApiModel):
records: List[CustomerRecord]
processed_at: datetime
total_count: int
@field_serializer('processed_at')
def serialize_datetime(self, dt: datetime, _info):
return dt.isoformat()
RecordList = TypeAdapter(List[CustomerRecord])
@validate_call
def ingest_data(
input_file: Annotated[str, Field(min_length=1)],
output_file: Annotated[str, Field(min_length=1)],
strict: bool = True
) -> dict:
"""Process CSV data and output JSON"""
print(f"Starting processing: {input_file}")
records = []
with open(input_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row_num, row in enumerate(reader, 1):
try:
# Pre‑process CSV row
processed_row = preprocess_row(row)
record = CustomerRecord.model_validate(processed_row)
records.append(record)
except Exception as e:
if strict:
raise ValueError(f"Row {row_num} validation failed: {e}")
else:
print(f"Warning: skipping row {row_num} – {e}")
continue
output = OutputPayload(
records=records,
processed_at=datetime.now(),
total_count=len(records)
)
Path(output_file).parent.mkdir(parents=True, exist_ok=True)
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(output.model_dump(by_alias=True, exclude_none=True), f, indent=2)
print(f"Processing complete: {len(records)} records → {output_file}")
return {"success": True, "processed": len(records), "output_file": output_file}
def preprocess_row(row: dict) -> dict:
"""Pre‑process a CSV row"""
processed = row.copy()
# Create immutable RecordId
processed['record_id'] = {'value': f"cust_{processed.get('id', 'unknown')}"}
# Convert types
if 'age' in processed:
processed['age'] = int(processed['age'])
if 'vip_status' in processed:
processed['vip_status'] = processed['vip_status'].lower() in ('true', '1', 'yes')
if 'signup_date' in processed:
processed['signup_date'] = datetime.fromisoformat(processed['signup_date'])
return processed
def main():
try:
settings = AppSettings()
print(f"Loaded config: input={settings.input_path}, output={settings.output_path}")
result = ingest_data(
input_file=settings.input_path,
output_file=settings.output_path,
strict=settings.strict_validation
)
print(f"✅ Success: {result}")
except Exception as e:
print(f"❌ Failure: {e}")
return 1
return 0
if __name__ == "__main__":
exit(main())This pipeline demonstrates how combining the twelve patterns yields a stable, type‑safe, and easily maintainable data‑processing service suitable for production environments.
Signed-in readers can open the original source through BestHub's protected redirect.
This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactand we will review it promptly.
Data STUDIO
Click to receive the "Python Study Handbook"; reply "benefit" in the chat to get it. Data STUDIO focuses on original data science articles, centered on Python, covering machine learning, data analysis, visualization, MySQL and other practical knowledge and project case studies.
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.
