Pydantic v2 Production Patterns: ConfigDict, валидация и сериализация для enterprise

from pydantic import BaseModel, ConfigDict, Field
from datetime import datetime
from uuid import UUID
 
class ProductionBaseModel(BaseModel):
    """Базовый класс для всех production моделей.
 
    Наследуйте от него для консистентной конфигурации."""
 
    model_config = ConfigDict(
        # Валидация
        strict=False,  # Обсудим trade-offs ниже
        validate_default=True,
        validate_assignment=False,  # Включать только если нужно
 
        # Дополнительные поля
        extra='forbid',  # Строго для API
 
        # ORM
        from_attributes=True,
 
        # Сериализация
        ser_json_timedelta='float',
        ser_json_bytes='base64',
        use_enum_values=True,
 
        # Производительность
        # revalidate_instances='never',  # Только если нужна максимальная скорость
    )
 
    # Стандартные поля
    id: UUID
    created_at: datetime
    updated_at: datetime | None = None

from pydantic import BaseModel
 
class Config(BaseModel):
    debug: bool
    timeout: int
 
# Проблемные кейсы без strict mode
config = Config(debug="false", timeout="30")
print(config.debug)  # True! Любая непустая строка = True
print(config.timeout)  # 30 (int) — OK, но неявно
 
# С потерей точности
class Payment(BaseModel):
    amount_cents: int
 
payment = Payment(amount_cents="100.99")
print(payment.amount_cents)  # 100 — потеряли .99 без ошибки!

# Стратегия 1: Глобальный strict для критичных моделей
class PaymentRequest(BaseModel):
    model_config = ConfigDict(strict=True)
 
    amount_cents: int
    currency: str
 
# Стратегия 2: Per-field strict для гибкости
class MixedModel(BaseModel):
    # Критичные поля — строгие
    price: Decimal = Field(strict=True)
    quantity: int = Field(strict=True)
 
    # User input — гибкий (API удобство)
    search_query: str  # Примет и int, приведёт к str
 
# Стратегия 3: Явная валидация в @field_validator
class SmartModel(BaseModel):
    value: int
 
    @field_validator('value', mode='before')
    @classmethod
    def parse_value(cls, v):
        # Контролируем приведение явно
        if isinstance(v, str):
            if not v.isdigit():
                raise ValueError(f'Invalid int string: {v}')
            return int(v)
        if isinstance(v, (int, float)):
            return int(v)
        raise ValueError(f'Cannot convert {type(v)} to int')

from pydantic import BaseModel, ConfigDict, ValidationError
 
# Опция 1: forbid (рекомендуем для API)
class StrictAPI(BaseModel):
    model_config = ConfigDict(extra='forbid')
 
    username: str
    email: str
 
try:
    user = StrictAPI(username="alice", email="a@ex.com", emial="typo@ex.com")
except ValidationError as e:
    print(e)
    # 1 validation error for StrictAPI
    # emial
    #   Extra inputs are not permitted
 
# Опция 2: allow (опасно, но иногда нужно)
class FlexibleAPI(BaseModel):
    model_config = ConfigDict(extra='allow')
 
    username: str
    email: str
 
user = FlexibleAPI(username="alice", email="a@ex.com", custom_field="data")
print(user.custom_field)  # "data" — сохранилось, но не типизировано!
 
# Опция 3: ignore (default в v2)
class DefaultAPI(BaseModel):
    username: str
    email: str
 
user = DefaultAPI(username="alice", email="a@ex.com", emial="typo")
# Создаётся без ошибки, emial молча игнорируется

# forbid — для публичных API (защита от опечаток клиентов)
class PublicAPIRequest(BaseModel):
    model_config = ConfigDict(extra='forbid')
 
# allow — для legacy API с обратной совместимостью
class LegacyAPIRequest(BaseModel):
    model_config = ConfigDict(extra='allow')
 
    # Известные поля
    user_id: int
    # Неизвестные поля сохраняются для backward compatibility
 
# ignore — для internal services (гибкость + не ломаем при добавлении полей)
class InternalMessage(BaseModel):
    model_config = ConfigDict(extra='ignore')

from pydantic import BaseModel, ConfigDict, Field
 
class Product(BaseModel):
    model_config = ConfigDict(validate_assignment=True)
 
    name: str
    price: float = Field(gt=0)  # Должна быть > 0
    stock: int = Field(ge=0)  # Должен быть >= 0
 
product = Product(name="Laptop", price=1000.0, stock=10)
 
# С validate_assignment=True — валидация при изменении
try:
    product.price = -100.0
except ValidationError:
    print("Cannot set negative price")  # Поймали!
 
# Без validate_assignment=True
class UnsafeProduct(BaseModel):
    name: str
    price: float = Field(gt=0)
 
unsafe = UnsafeProduct(name="Laptop", price=1000.0)
unsafe.price = -100.0  # Принято без проверки!
print(unsafe.price)  # -100.0 — инвариант нарушен

# Лучше чем validate_assignment — сделать immutable
class ImmutableProduct(BaseModel):
    model_config = ConfigDict(frozen=True)
 
    name: str
    price: float = Field(gt=0)
 
product = ImmutableProduct(name="Laptop", price=1000.0)
 
# Любая попытка изменить — ошибка
try:
    product.price = 1500.0
except ValidationError:
    print("Cannot modify frozen model")
 
# Для изменений — создаём новый объект
updated = product.model_copy(update={"price": 1500.0})  # Валидация при создании

# Модель инвойса
class Invoice(BaseModel):
    amount: Decimal
    currency: str
 
# Европейские клиенты отправляют: "1.234,56" (EU format)
# Американские клиенты: "1,234.56" (US format)
 
# С strict=False:
Invoice(amount="1.234,56", currency="EUR")  # Decimal('1.234') ← потеря .56!
Invoice(amount="1,234.56", currency="USD")  # Decimal('1.23456') ← неправильный парсинг!

from decimal import Decimal
from pydantic import field_validator, ValidationInfo
 
class Invoice(BaseModel):
    model_config = ConfigDict(strict=True)  # Только точные типы
 
    amount: Decimal
    currency: str
 
    @field_validator('amount', mode='before')
    @classmethod
    def parse_amount_with_locale(cls, v, info: ValidationInfo):
        """Парсинг цены с учётом locale клиента."""
        if isinstance(v, str):
            # Получаем locale из context (передаём из API endpoint)
            locale = info.context.get('locale', 'en_US') if info.context else 'en_US'
 
            if locale.startswith('de') or locale.startswith('fr'):  # EU format
                # "1.234,56" → "1234.56"
                v = v.replace('.', '').replace(',', '.')
            else:  # US format
                # "1,234.56" → "1234.56"
                v = v.replace(',', '')
 
            return Decimal(v)
 
        return v
 
# В API endpoint передаём locale
@app.post("/invoices")
def create_invoice(invoice_data: dict, locale: str = Header(default="en_US")):
    invoice = Invoice.model_validate(
        invoice_data,
        context={'locale': locale}
    )
    return invoice

class FeatureFlags(BaseModel):
    new_checkout: bool
    beta_ui: bool
    debug_mode: bool
 
# В Redis хранились строки "true"/"false"
flags_data = redis.hgetall("feature_flags")
# {'new_checkout': 'false', 'beta_ui': 'false', 'debug_mode': 'false'}
 
flags = FeatureFlags(**flags_data)
print(flags.new_checkout)  # True ← ЧТО?!
print(flags.beta_ui)       # True
print(flags.debug_mode)    # True

class FeatureFlags(BaseModel):
    model_config = ConfigDict(strict=True)  # Требуем точные типы
 
    new_checkout: bool
    beta_ui: bool
    debug_mode: bool
 
# В коде загрузки явно конвертируем
def load_flags(redis_client) -> FeatureFlags:
    raw_data = redis_client.hgetall("feature_flags")
 
    # Явное приведение строк в bool
    parsed_data = {
        k: v.lower() == 'true' if isinstance(v, str) else v
        for k, v in raw_data.items()
    }
 
    return FeatureFlags(**parsed_data)

class FeatureFlags(BaseModel):
    new_checkout: bool
    beta_ui: bool
    debug_mode: bool
 
    @model_validator(mode='before')
    @classmethod
    def parse_string_bools(cls, data):
        """Конвертируем строковые boolean из Redis."""
        if isinstance(data, dict):
            for key, value in data.items():
                if isinstance(value, str):
                    if value.lower() in ('true', '1', 'yes', 'on'):
                        data[key] = True
                    elif value.lower() in ('false', '0', 'no', 'off'):
                        data[key] = False
        return data

class DataRecord(BaseModel):
    id: int
    timestamp: datetime
    values: list[float]
    metadata: dict
 
# Обработка 500k записей
records = [DataRecord(**row) for row in csv_reader]  # Peak memory: 4.2GB

# Вариант 1: model_construct для trusted данных
def process_csv_fast(csv_path: str):
    # Валидируем структуру файла один раз
    with open(csv_path) as f:
        reader = csv.DictReader(f)
        headers = reader.fieldnames
 
        expected = set(DataRecord.model_fields.keys())
        if not expected.issubset(headers):
            raise ValueError(f"Invalid CSV structure")
 
        # Создаём без валидации (данные проверены на уровне схемы)
        records = [
            DataRecord.model_construct(**row)
            for row in reader
        ]
 
    return records  # Peak memory: 890MB (в 4.7x меньше)
 
# Вариант 2: Batch processing с явной очисткой
import gc
 
def process_csv_batched(csv_path: str, batch_size: int = 10_000):
    results = []
 
    with open(csv_path) as f:
        reader = csv.DictReader(f)
        batch = []
 
        for i, row in enumerate(reader):
            batch.append(DataRecord(**row))
 
            if len(batch) >= batch_size:
                # Обрабатываем batch
                results.extend(process_batch(batch))
 
                # Явно очищаем память
                batch.clear()
                gc.collect()
 
    return results  # Peak memory: 1.2GB (в 3.5x меньше)

from pydantic import BaseModel, Field
from typing import Annotated
from datetime import datetime
 
class User(BaseModel):
    # Строки
    username: str = Field(min_length=3, max_length=20, pattern=r'^[a-zA-Z0-9_]+$')
    email: str = Field(pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
 
    # Числа
    age: int = Field(ge=18, le=120)
    rating: float = Field(gt=0, le=5)
 
    # Даты
    birth_date: datetime = Field(le=datetime.now())  # Не в будущем
 
    # Коллекции
    tags: list[str] = Field(min_length=1, max_length=10)
    metadata: dict = Field(default_factory=dict)
 
# Переиспользуемые типы через Annotated
Username = Annotated[str, Field(min_length=3, max_length=20, pattern=r'^[a-zA-Z0-9_]+$')]
Email = Annotated[str, Field(pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')]
Age = Annotated[int, Field(ge=18, le=120)]
 
class CompactUser(BaseModel):
    username: Username
    email: Email
    age: Age

from pydantic import BaseModel, field_validator
from typing import Any
 
class Product(BaseModel):
    name: str
    price: float
    discount_percent: float = 0.0
 
    @field_validator('price', mode='before')
    @classmethod
    def parse_price(cls, v: Any) -> Any:
        """Парсинг цены из разных форматов ДО type checking."""
        # mode='before' получает сырые данные (может быть str, int, float, etc.)
        if isinstance(v, str):
            # Убираем символы валюты и пробелы
            cleaned = v.replace('$', '').replace('€', '').replace(' ', '').replace(',', '')
            return float(cleaned)
        return v
 
    @field_validator('price', mode='after')
    @classmethod
    def round_price(cls, v: float) -> float:
        """Округление ПОСЛЕ type checking и Field constraints."""
        # mode='after' получает уже проверенный float
        return round(v, 2)
 
    @field_validator('discount_percent', mode='after')
    @classmethod
    def validate_discount(cls, v: float) -> float:
        if not 0 <= v <= 100:
            raise ValueError('Discount must be between 0 and 100')
        return v
 
# Примеры использования
product = Product(name="Laptop", price="$1,299.99", discount_percent=10)
print(product.price)  # 1299.99
 
# Валидация сработает
try:
    Product(name="Laptop", price="$1,299.99", discount_percent=150)
except ValidationError as e:
    print(e)  # Discount must be between 0 and 100

from pydantic import BaseModel, field_validator, model_validator
 
# ❌ Неправильно: пытаемся сравнить поля в field_validator
class WrongOrder(BaseModel):
    start_date: datetime
    end_date: datetime
 
    @field_validator('end_date', mode='after')
    @classmethod
    def validate_end_after_start(cls, v: datetime) -> datetime:
        # Проблема: доступа к start_date здесь нет!
        # Это field_validator, он видит только одно поле
        return v
 
# ✅ Правильно: используем model_validator для cross-field validation
class CorrectOrder(BaseModel):
    start_date: datetime
    end_date: datetime
 
    @model_validator(mode='after')
    def validate_dates_order(self):
        # Теперь видим все поля модели
        if self.end_date < self.start_date:
            raise ValueError('end_date must be after start_date')
        return self

from typing import Annotated
from pydantic import BaseModel, AfterValidator, BeforeValidator, WrapValidator
import re
 
# Простая валидация через функцию
def validate_phone(v: str) -> str:
    digits = re.sub(r'\D', '', v)
    if len(digits) != 11 or not digits.startswith('7'):
        raise ValueError('Phone must be 11 digits starting with 7')
    return f'+7 ({digits[1:4]}) {digits[4:7]}-{digits[7:9]}-{digits[9:]}'
 
PhoneNumber = Annotated[str, AfterValidator(validate_phone)]
 
# Продвинутый кейс: WrapValidator для полного контроля
from pydantic_core import core_schema
from pydantic import GetCoreSchemaHandler
 
class StrictEmail(str):
    """Email с валидацией на Pydantic core уровне (Rust)."""
 
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler: GetCoreSchemaHandler):
        # Получаем стандартную схему для str
        str_schema = handler(str)
 
        return core_schema.no_info_after_validator_function(
            cls._validate,
            str_schema,
            serialization=core_schema.plain_serializer_function_ser_schema(
                lambda x: str(x)
            )
        )
 
    @classmethod
    def _validate(cls, v: str) -> "StrictEmail":
        # Валидация email
        if '@' not in v or '.' not in v.split('@')[1]:
            raise ValueError('Invalid email format')
 
        # Дополнительная проверка: нет одноразовых email
        disposable_domains = ['temp-mail.org', '10minutemail.com']
        domain = v.split('@')[1]
        if domain in disposable_domains:
            raise ValueError(f'Disposable email domains not allowed: {domain}')
 
        return cls(v)
 
class User(BaseModel):
    phone: PhoneNumber
    email: StrictEmail
 
# Использование
user = User(phone="8-900-123-45-67", email="user@example.com")
print(user.phone)  # +7 (900) 123-45-67

from pydantic import BaseModel, model_validator, ValidationInfo
 
# ✅ Правильно: используем context для runtime данных БЕЗ I/O
class Article(BaseModel):
    title: str
    author_id: int
    is_published: bool
 
    @model_validator(mode='after')
    def check_permissions(self, info: ValidationInfo):
        """Проверка прав на основе context (без I/O)."""
        if not info.context:
            return self
 
        current_user_id = info.context.get('current_user_id')
        is_admin = info.context.get('is_admin', False)
 
        if self.is_published:
            if not is_admin and self.author_id != current_user_id:
                raise ValueError('Only author or admin can publish')
 
        return self
 
# В FastAPI endpoint
@app.post("/articles")
async def create_article(
    article_data: dict,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    # Валидация схемы + permissions через context
    article = Article.model_validate(
        article_data,
        context={
            "current_user_id": current_user.id,
            "is_admin": current_user.is_admin
        }
    )
 
    # I/O операции после валидации схемы
    db.add(Article(**article.model_dump()))
    db.commit()

# ❌ Антипаттерн: I/O в валидаторе
class OrderCreate(BaseModel):
    product_id: int
 
    @field_validator('product_id')
    @classmethod
    def product_exists(cls, v: int) -> int:
        # Блокирующий I/O в валидаторе!
        product = db.query(Product).get(v)  # ← НЕ ДЕЛАЙТЕ ТАК
        if not product:
            raise ValueError(f'Product {v} not found')
        return v
 
# ✅ Правильно: I/O через dependency injection
from fastapi import Depends, HTTPException
 
async def validate_product_exists(
    product_id: int,
    db: Session = Depends(get_db)
) -> Product:
    """Dependency для проверки существования продукта."""
    product = await db.get(Product, product_id)
    if not product:
        raise HTTPException(404, f"Product {product_id} not found")
    if not product.is_available:
        raise HTTPException(400, f"Product {product_id} is not available")
    return product
 
@app.post("/orders")
async def create_order(
    order_data: OrderCreate,  # Pydantic валидирует схему
    product: Product = Depends(validate_product_exists)  # I/O отдельно
):
    # Теперь product уже проверен и загружен
    order = Order(product_id=product.id, ...)
    return order

from pydantic import BaseModel, ValidationError
 
class User(BaseModel):
    email: str
    age: int
 
try:
    User(email="invalid", age="not a number")
except ValidationError as e:
    print(e)
 
# Output:
# 2 validation errors for User
# email
#   value is not a valid email address (type=value_error.email)
# age
#   value is not a valid integer (type=type_error.integer)

from pydantic import BaseModel, ValidationError, Field
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from typing import Any
 
app = FastAPI()
 
class ErrorResponse(BaseModel):
    """Стандартный формат ошибок API."""
    error_code: str
    message: str
    details: list[dict[str, Any]] | None = None
 
class FieldError(BaseModel):
    """Ошибка конкретного поля."""
    field: str
    message: str
    value: Any = None
 
def format_validation_error(exc: ValidationError) -> ErrorResponse:
    """Преобразует Pydantic ValidationError в user-friendly формат."""
    field_errors = []
 
    for error in exc.errors():
        # error['loc'] — путь к полю, например ('email',) или ('address', 'city')
        field_path = '.'.join(str(loc) for loc in error['loc'])
 
        # Человекочитаемые сообщения
        message = get_user_friendly_message(error)
 
        field_errors.append(FieldError(
            field=field_path,
            message=message,
            value=error.get('input')
        ))
 
    return ErrorResponse(
        error_code="VALIDATION_ERROR",
        message="Проверьте правильность заполнения полей",
        details=[err.model_dump() for err in field_errors]
    )
 
def get_user_friendly_message(error: dict) -> str:
    """Маппинг Pydantic error types в понятные сообщения."""
    error_type = error['type']
 
    messages = {
        'string_type': 'Должно быть текстом',
        'int_type': 'Должно быть целым числом',
        'float_type': 'Должно быть числом',
        'value_error.email': 'Неверный формат email',
        'value_error.url': 'Неверный формат URL',
        'string_too_short': lambda ctx: f'Минимум {ctx.get("min_length")} символов',
        'string_too_long': lambda ctx: f'Максимум {ctx.get("max_length")} символов',
        'string_pattern_mismatch': 'Неверный формат',
        'greater_than': lambda ctx: f'Должно быть больше {ctx.get("gt")}',
        'greater_than_equal': lambda ctx: f'Должно быть не меньше {ctx.get("ge")}',
        'less_than': lambda ctx: f'Должно быть меньше {ctx.get("lt")}',
        'less_than_equal': lambda ctx: f'Должно быть не больше {ctx.get("le")}',
        'missing': 'Обязательное поле',
    }
 
    msg = messages.get(error_type, 'Некорректное значение')
 
    # Если сообщение — функция, вызываем с контекстом
    if callable(msg):
        return msg(error.get('ctx', {}))
 
    return msg
 
# Использование в FastAPI
@app.exception_handler(ValidationError)
async def validation_exception_handler(request, exc: ValidationError):
    """Глобальный handler для ValidationError."""
    error_response = format_validation_error(exc)
 
    return JSONResponse(
        status_code=422,
        content=error_response.model_dump()
    )
 
@app.post("/users")
def create_user(user: User):
    # Валидация происходит автоматически через FastAPI
    return user
 
# Пример ответа при ошибке:
# {
#   "error_code": "VALIDATION_ERROR",
#   "message": "Проверьте правильность заполнения полей",
#   "details": [
#     {
#       "field": "email",
#       "message": "Неверный формат email",
#       "value": "invalid"
#     },
#     {
#       "field": "age",
#       "message": "Должно быть целым числом",
#       "value": "not a number"
#     }
#   ]
# }

import logging
import structlog
from prometheus_client import Counter
 
logger = structlog.get_logger()
 
# Prometheus метрики
validation_errors_total = Counter(
    'validation_errors_total',
    'Total validation errors',
    ['model_name', 'field', 'error_type']
)
 
def log_validation_error(exc: ValidationError, model_name: str, context: dict = None):
    """Структурированное логирование ошибок валидации."""
    for error in exc.errors():
        field_path = '.'.join(str(loc) for loc in error['loc'])
 
        # Структурированный лог
        logger.warning(
            "validation_error",
            model=model_name,
            field=field_path,
            error_type=error['type'],
            input_value=error.get('input'),
            context=context or {}
        )
 
        # Метрики для мониторинга
        validation_errors_total.labels(
            model_name=model_name,
            field=field_path,
            error_type=error['type']
        ).inc()
 
# Использование
@app.post("/orders")
def create_order(order_data: dict, user_id: int = Depends(get_current_user_id)):
    try:
        order = Order(**order_data)
    except ValidationError as e:
        # Логируем для аналитики
        log_validation_error(
            e,
            model_name="Order",
            context={
                "user_id": user_id,
                "endpoint": "/orders",
                "ip": request.client.host
            }
        )
 
        # Возвращаем user-friendly ошибку
        raise HTTPException(
            status_code=422,
            detail=format_validation_error(e).model_dump()
        )
 
    return order

from pydantic import BaseModel, ValidationError
from typing import TypeVar, Generic
 
T = TypeVar('T')
 
class PartialResult(Generic[T]):
    """Результат частичной валидации."""
    valid: list[T]
    invalid: list[dict]
 
def validate_partial(
    model: type[BaseModel],
    data: list[dict]
) -> PartialResult:
    """Валидирует список, собирая и валидные и невалидные записи."""
    valid = []
    invalid = []
 
    for i, row in enumerate(data):
        try:
            obj = model(**row)
            valid.append(obj)
        except ValidationError as e:
            invalid.append({
                'index': i,
                'data': row,
                'errors': e.errors()
            })
 
    return PartialResult(valid=valid, invalid=invalid)
 
# Пример: импорт CSV с частичной валидацией
@app.post("/import/users")
def import_users(file: UploadFile):
    # Читаем CSV
    rows = list(csv.DictReader(file.file))
 
    # Валидируем частично
    result = validate_partial(User, rows)
 
    # Сохраняем валидные
    db.bulk_insert_mappings(User, [u.model_dump() for u in result.valid])
 
    # Возвращаем отчёт
    return {
        "imported": len(result.valid),
        "failed": len(result.invalid),
        "errors": result.invalid[:10]  # Первые 10 ошибок
    }

from pydantic import BaseModel, ValidationError, field_validator
 
class UserRegistration(BaseModel):
    email: str
    phone: str
 
    @field_validator('email', mode='before')
    @classmethod
    def normalize_email(cls, v):
        """Нормализация email с автокоррекцией частых ошибок."""
        if isinstance(v, str):
            v = v.strip().lower()
 
            # Автокоррекция частых опечаток
            typos = {
                'gmial.com': 'gmail.com',
                'gmai.com': 'gmail.com',
                'yahooo.com': 'yahoo.com',
                'hotmial.com': 'hotmail.com'
            }
 
            for typo, correct in typos.items():
                v = v.replace(typo, correct)
 
        return v
 
    @field_validator('phone', mode='before')
    @classmethod
    def normalize_phone(cls, v):
        """Нормализация телефона — убираем всё кроме цифр."""
        if isinstance(v, str):
            # Убираем пробелы, тире, скобки
            v = ''.join(filter(str.isdigit, v))
 
            # Добавляем +7 для российских номеров
            if len(v) == 10:
                v = f'+7{v}'
            elif len(v) == 11 and v.startswith('7'):
                v = f'+{v}'
            elif len(v) == 11 and v.startswith('8'):
                v = f'+7{v[1:]}'
 
        return v
 
# Пример работы автокоррекции
user = UserRegistration(
    email="  User@GMIal.COM  ",  # → user@gmail.com
    phone="8 (900) 123-45-67"     # → +79001234567
)

# Query 1: Top 10 полей с ошибками
topk(10,
  sum by (field) (
    rate(validation_errors_total[5m])
  )
)
 
# Query 2: Распределение типов ошибок
sum by (error_type) (
  rate(validation_errors_total[1h])
)
 
# Query 3: Spike detection (аномальный рост ошибок)
rate(validation_errors_total[5m]) > 2 * rate(validation_errors_total[1h] offset 1d)
 
# Alert rule: слишком много ошибок валидации
ALERT HighValidationErrorRate
  IF rate(validation_errors_total[5m]) > 100
  FOR 5m
  ANNOTATIONS {
    summary = "High validation error rate",
    description = "{{ $value }} validation errors per second"
  }

from pydantic import BaseModel, model_validator, Field
from typing import Literal
 
class PaymentMethod(BaseModel):
    """Модель с условной валидацией на основе типа платежа."""
    type: Literal["card", "bank_transfer", "crypto"]
 
    # Поля для card
    card_number: str | None = None
    cvv: str | None = None
    expiry: str | None = None
 
    # Поля для bank_transfer
    iban: str | None = None
    swift: str | None = None
 
    # Поля для crypto
    wallet_address: str | None = None
    network: str | None = None
 
    @model_validator(mode='after')
    def validate_payment_fields(self):
        """Проверяем, что заполнены нужные поля в зависимости от типа."""
        if self.type == "card":
            if not all([self.card_number, self.cvv, self.expiry]):
                raise ValueError(
                    "Card payment requires card_number, cvv, and expiry"
                )
            # Валидация карты
            if len(self.card_number.replace(" ", "")) != 16:
                raise ValueError("Card number must be 16 digits")
 
        elif self.type == "bank_transfer":
            if not all([self.iban, self.swift]):
                raise ValueError(
                    "Bank transfer requires iban and swift"
                )
 
        elif self.type == "crypto":
            if not all([self.wallet_address, self.network]):
                raise ValueError(
                    "Crypto payment requires wallet_address and network"
                )
 
        return self
 
# Использование
card_payment = PaymentMethod(
    type="card",
    card_number="1234 5678 9012 3456",
    cvv="123",
    expiry="12/25"
)
 
crypto_payment = PaymentMethod(
    type="crypto",
    wallet_address="0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb1",
    network="ethereum"
)

from pydantic import BaseModel, field_validator, ValidationInfo
from typing import Any
 
class RegistrationRequest(BaseModel):
    """Регистрация с динамическими правилами валидации."""
    username: str
    email: str
    password: str
    age: int
 
    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str, info: ValidationInfo) -> str:
        """Валидация пароля с учётом feature flags."""
        # Получаем feature flags из context
        if not info.context:
            return v
 
        strict_mode = info.context.get('strict_password_policy', False)
 
        if strict_mode:
            # Строгие требования
            if len(v) < 12:
                raise ValueError('Password must be at least 12 characters (strict mode)')
 
            if not any(c.isupper() for c in v):
                raise ValueError('Password must contain uppercase letter')
 
            if not any(c.isdigit() for c in v):
                raise ValueError('Password must contain digit')
 
            if not any(c in '!@#$%^&*' for c in v):
                raise ValueError('Password must contain special character')
        else:
            # Базовые требования
            if len(v) < 8:
                raise ValueError('Password must be at least 8 characters')
 
        return v
 
# В API endpoint передаём feature flags
@app.post("/register")
def register(
    data: dict,
    strict_mode: bool = Depends(get_strict_mode_flag)
):
    user = RegistrationRequest.model_validate(
        data,
        context={'strict_password_policy': strict_mode}
    )
    return user

from pydantic import BaseModel, model_validator
from datetime import datetime, timedelta
 
class Subscription(BaseModel):
    """Подписка с каскадной валидацией."""
    plan: Literal["free", "basic", "premium"]
    start_date: datetime
    end_date: datetime | None = None
    auto_renew: bool = False
    promo_code: str | None = None
 
    @model_validator(mode='after')
    def validate_subscription_rules(self):
        """Каскадная проверка правил подписки."""
        # Правило 1: end_date должна быть после start_date
        if self.end_date and self.end_date <= self.start_date:
            raise ValueError('end_date must be after start_date')
 
        # Правило 2: Free план не может быть auto_renew
        if self.plan == "free" and self.auto_renew:
            raise ValueError('Free plan cannot have auto_renew enabled')
 
        # Правило 3: Promo code только для платных планов
        if self.plan == "free" and self.promo_code:
            raise ValueError('Free plan cannot use promo codes')
 
        # Правило 4: Premium требует минимум 1 год подписки
        if self.plan == "premium" and self.end_date:
            duration = self.end_date - self.start_date
            if duration < timedelta(days=365):
                raise ValueError('Premium plan requires minimum 1 year subscription')
 
        return self

from pydantic import BaseModel, field_validator
import httpx
from functools import lru_cache
 
class OrderCreate(BaseModel):
    """Заказ с проверкой через внешние сервисы."""
    product_id: str
    quantity: int
    shipping_address: str
    promo_code: str | None = None
 
    @field_validator('promo_code')
    @classmethod
    def validate_promo_code(cls, v: str | None) -> str | None:
        """Валидация промокода через promo-service."""
        if v is None:
            return v
 
        # ⚠️ Внимание: I/O в валидаторе — используйте осторожно!
        # Лучше проверять промокод в endpoint, а не здесь
 
        # Если всё же нужно здесь, используйте кэш
        if not is_promo_code_valid(v):
            raise ValueError(f'Invalid promo code: {v}')
 
        return v
 
@lru_cache(maxsize=1000)
def is_promo_code_valid(code: str) -> bool:
    """Проверка промокода с кэшированием."""
    try:
        # Синхронный HTTP запрос к promo-service
        response = httpx.get(
            f"http://promo-service/validate/{code}",
            timeout=1.0  # Короткий timeout чтобы не блокировать
        )
        return response.status_code == 200
    except httpx.TimeoutException:
        # Fail open: если сервис недоступен, пропускаем
        return True
    except Exception:
        return False

from fastapi import Depends, HTTPException
 
# ✅ Правильно: I/O через dependency injection
class OrderCreate(BaseModel):
    """Заказ без I/O в валидаторе."""
    product_id: str
    quantity: int
    shipping_address: str
    promo_code: str | None = None
 
async def validate_promo_code(
    order: OrderCreate,
    promo_service: PromoService = Depends(get_promo_service)
) -> OrderCreate:
    """Dependency для async проверки промокода."""
    if order.promo_code:
        is_valid = await promo_service.validate(order.promo_code)
        if not is_valid:
            raise HTTPException(400, f"Invalid promo code: {order.promo_code}")
 
    return order
 
@app.post("/orders")
async def create_order(
    order: OrderCreate = Depends(validate_promo_code)  # Валидация через dependency
):
    # Здесь order уже провалидирован (и схема, и промокод)
    db.add(order)
    return order

from pydantic import BaseModel, field_validator, model_validator
from typing import Any
 
class ComplexForm(BaseModel):
    """Форма с многоэтапной валидацией."""
    email: str
    password: str
    password_confirm: str
    terms_accepted: bool
 
    # Этап 1: Field-level validation (проверка формата)
    @field_validator('email')
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        """Проверка формата email."""
        if '@' not in v or '.' not in v.split('@')[1]:
            raise ValueError('Invalid email format')
        return v.lower()
 
    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        """Проверка силы пароля."""
        if len(v) < 8:
            raise ValueError('Password too short')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        return v
 
    # Этап 2: Model-level validation (проверка связей между полями)
    @model_validator(mode='after')
    def validate_passwords_match(self):
        """Проверка совпадения паролей."""
        if self.password != self.password_confirm:
            raise ValueError('Passwords do not match')
        return self
 
    @model_validator(mode='after')
    def validate_terms_accepted(self):
        """Проверка принятия условий."""
        if not self.terms_accepted:
            raise ValueError('You must accept terms and conditions')
        return self
 
# Этап 3: Business logic validation (в endpoint)
@app.post("/register")
async def register(form: ComplexForm):
    # Проверяем уникальность email (I/O операция)
    existing_user = await db.users.find_one({"email": form.email})
    if existing_user:
        raise HTTPException(400, "Email already registered")
 
    # Создаём пользователя
    user = await create_user(form)
    return user

from pydantic import BaseModel, model_validator
from typing import Literal
 
class UserV1(BaseModel):
    """Версия 1: одно поле для имени."""
    version: Literal["v1"] = "v1"
    name: str
    email: str
 
class UserV2(BaseModel):
    """Версия 2: раздельные first_name и last_name."""
    version: Literal["v2"] = "v2"
    first_name: str
    last_name: str
    email: str
 
    @model_validator(mode='before')
    @classmethod
    def migrate_from_v1(cls, data: Any) -> Any:
        """Автоматическая миграция из v1."""
        if isinstance(data, dict):
            # Определяем v1 по наличию поля 'name'
            if 'name' in data and 'first_name' not in data:
                parts = data['name'].split(' ', 1)
                data['first_name'] = parts[0]
                data['last_name'] = parts[1] if len(parts) > 1 else ''
                data.pop('name', None)
                data['version'] = 'v2'
 
        return data
 
# API принимает обе версии
@app.post("/users")
def create_user(user_data: dict):
    # Пробуем v2 (с автомиграцией из v1)
    try:
        user = UserV2(**user_data)
    except ValidationError:
        # Fallback на v1 (если данные совсем не подходят)
        user = UserV1(**user_data)
 
    # Внутренняя логика работает только с v2
    if isinstance(user, UserV1):
        user = convert_v1_to_v2(user)
 
    db.add(user)
    return user

from pydantic import BaseModel, Field
from typing import Optional
 
class UserUpdate(BaseModel):
    """Модель для частичного обновления пользователя."""
    username: str | None = None
    email: str | None = None
    bio: str | None = None
 
    @model_validator(mode='after')
    def at_least_one_field(self):
        """Проверка, что хотя бы одно поле передано."""
        if all(v is None for v in [self.username, self.email, self.bio]):
            raise ValueError('At least one field must be provided')
        return self
 
@app.patch("/users/{user_id}")
def update_user(user_id: int, update: UserUpdate):
    # Получаем существующего пользователя
    user = db.query(User).get(user_id)
 
    # Обновляем только переданные поля
    update_data = update.model_dump(exclude_unset=True)  # Только заполненные поля
 
    for field, value in update_data.items():
        setattr(user, field, value)
 
    db.commit()
    return user

from pydantic import BaseModel, field_serializer
from datetime import datetime
from decimal import Decimal
from uuid import UUID
 
class Transaction(BaseModel):
    id: UUID
    amount: Decimal
    currency: str
    created_at: datetime
    description: str | None = None
 
    @field_serializer('id')
    def serialize_uuid(self, value: UUID) -> str:
        """UUID → строка для JSON."""
        return str(value)
 
    @field_serializer('amount')
    def serialize_amount(self, value: Decimal, _info) -> str:
        """Decimal → строка с 2 знаками для точности."""
        return f"{value:.2f}"
 
    @field_serializer('created_at')
    def serialize_datetime(self, value: datetime) -> str:
        """Datetime → ISO 8601."""
        return value.isoformat()
 
    @field_serializer('description')
    def serialize_description(self, value: str | None) -> str:
        """None → пустая строка для клиента."""
        return value if value else ""
 
# Результат
tx = Transaction(
    id=UUID("123e4567-e89b-12d3-a456-426614174000"),
    amount=Decimal("100.5"),
    currency="USD",
    created_at=datetime(2025, 12, 4, 10, 30),
    description=None
)
 
print(tx.model_dump())
# {
#     'id': '123e4567-e89b-12d3-a456-426614174000',
#     'amount': '100.50',
#     'currency': 'USD',
#     'created_at': '2025-12-04T10:30:00',
#     'description': ''
# }

from pydantic import BaseModel, model_serializer
 
class User(BaseModel):
    id: int
    username: str
    email: str
    password_hash: str
    is_active: bool
    roles: list[str]
 
    @model_serializer
    def serialize_model(self):
        """Кастомная сериализация: исключаем password_hash, добавляем computed поля."""
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            # password_hash намеренно пропущен
            'status': 'active' if self.is_active else 'inactive',
            'permissions': self._compute_permissions()
        }
 
    def _compute_permissions(self) -> list[str]:
        """Вычисление permissions на основе roles."""
        permissions_map = {
            'admin': ['read', 'write', 'delete', 'manage_users'],
            'editor': ['read', 'write'],
            'viewer': ['read']
        }
        permissions = set()
        for role in self.roles:
            permissions.update(permissions_map.get(role, []))
        return sorted(permissions)
 
user = User(
    id=1,
    username="alice",
    email="alice@example.com",
    password_hash="$2b$12$KIXqz...",
    is_active=True,
    roles=["admin", "editor"]
)
 
print(user.model_dump())
# {
#     'id': 1,
#     'username': 'alice',
#     'email': 'alice@example.com',
#     'status': 'active',
#     'permissions': ['delete', 'manage_users', 'read', 'write']
# }

from pydantic import BaseModel, Field, ConfigDict
from pydantic.alias_generators import to_camel
 
# Автоматическое преобразование всех полей
class AutoCamelCaseModel(BaseModel):
    model_config = ConfigDict(
        alias_generator=to_camel,
        populate_by_name=True  # Принимаем и snake_case и camelCase на вход
    )
 
    user_id: int
    first_name: str
    is_active: bool
 
# Использование
user = AutoCamelCaseModel(user_id=1, first_name="Alice", is_active=True)
print(user.model_dump(by_alias=True))
# {'userId': 1, 'firstName': 'Alice', 'isActive': True}
 
# Ручное управление алиасами для сложных кейсов
class ManualAliasModel(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
 
    user_id: int = Field(serialization_alias="userId")
    first_name: str = Field(
        validation_alias="firstName",  # На вход принимаем firstName
        serialization_alias="first_name"  # На выход отдаём first_name
    )
 
# Поддержка legacy API
from pydantic import AliasChoices
 
class LegacyCompatibleModel(BaseModel):
    user_id: int = Field(
        validation_alias=AliasChoices('userId', 'user_id', 'id'),  # Принимаем все варианты
        serialization_alias='userId'  # Отдаём только один
    )
 
# Все эти варианты работают
m1 = LegacyCompatibleModel.model_validate({'userId': 1})
m2 = LegacyCompatibleModel.model_validate({'user_id': 1})
m3 = LegacyCompatibleModel.model_validate({'id': 1})
 
# Но сериализация всегда одинаковая
print(m1.model_dump(by_alias=True))  # {'userId': 1}