28 моделей Pydantic v1, шесть вечеров и три пожара на стенде

# Pydantic v1
class JWTPayload(BaseModel):
    user_id: int
    email: str
    roles: List[str]
    exp: int
    iat: int
 
# Метрики до миграции
Latency (P95): 12ms на валидацию токена
CPU: 18% utilization на валидацию
Throughput: ~2000 req/sec до CPU throttling

orders = [Order(**row) for row in csv_reader]  # Пиковое потребление: 1.2GB

# 1000 пользователей с профилями и правами
users = [UserResponse.from_orm(u) for u in db_users]
serialized = [u.dict() for u in users]  # ~450ms

class Order(BaseModel):
    product_id: int
    quantity: int
    price: Decimal
 
# Обычное создание (с валидацией)
order = Order(product_id="123", quantity="5", price="99.99")
# 1. Валидация типов: "123" -> int, "5" -> int
# 2. Валидация constraints (если есть)
# 3. Создание объекта
# Время: ~10-50 μs на объект (зависит от сложности модели)

# Создание без валидации
order = Order.model_construct(product_id=123, quantity=5, price=Decimal("99.99"))
# 1. Просто создаёт объект с переданными значениями
# 2. НЕТ проверки типов
# 3. НЕТ валидаторов
# Время: ~1-2 μs на объект (в 5-25 раз быстрее)

# CSV с 50,000 заказов от внутренней системы
# Формат гарантирован (генерируется нашим же кодом)
 
# ❌ Медленно: валидация каждой строки
orders = [Order(**row) for row in csv_rows]  # ~2.5 секунды + 1.2GB RAM
 
# ✅ Быстро: валидируем только структуру файла
def validate_csv_structure(headers):
    expected = set(Order.model_fields.keys())
    if not expected.issubset(headers):
        raise ValueError(f"Invalid CSV: missing {expected - set(headers)}")
 
validate_csv_structure(csv_headers)  # Один раз для файла
orders = [Order.model_construct(**row) for row in csv_rows]  # ~0.3 секунды + 380MB RAM

# Скрипт для поиска всех BaseModel (rg = ripgrep, быстрая альтернатива grep)
rg "class \w+\(BaseModel\)" --type py -c
 
# Результат:
API Gateway: 8 моделей
Order Service: 12 моделей
User Service: 6 моделей
Shared libs: 2 модели
Итого: 28 моделей

# check_dependencies.py
import ast
import sys
 
class ValidatorVisitor(ast.NodeVisitor):
    """Обходит AST Python-файла и собирает статистику использования Pydantic v1."""
 
    def __init__(self):
        self.validators = []        # Список моделей с @validator
        self.config_classes = []    # Список моделей с вложенным class Config
 
    def visit_FunctionDef(self, node):
        """Вызывается для каждой функции/метода в коде."""
        # Проверяем декораторы метода
        for decorator in node.decorator_list:
            # Если нашли @validator, это v1 API, в v2 нужен @field_validator
            if isinstance(decorator, ast.Name) and decorator.id == 'validator':
                self.validators.append(node.name)
 
    def visit_ClassDef(self, node):
        """Вызывается для каждого класса в коде."""
        # Проверяем тело класса на наличие вложенного class Config
        for item in node.body:
            if isinstance(item, ast.ClassDef) and item.name == 'Config':
                # class Config -> model_config = ConfigDict() в v2
                self.config_classes.append(node.name)
        self.generic_visit(node)  # Продолжаем обход вложенных узлов
 
# Запускаем на всех Python-файлах проекта, собираем статистику
# Результаты анализа:
# 18 моделей с @validator          ← нужно менять на @field_validator
# 14 моделей с классом Config       ← нужно менять на model_config = ConfigDict()
# 8 моделей с orm_mode = True       ← нужно менять на from_attributes = True
# 3 модели с JSON encoders          ← нужно переписывать на @field_serializer

# migrate_pydantic.py
import libcst as cst
from libcst import matchers as m
 
class PydanticV2Transformer(cst.CSTTransformer):
    """Автоматическая миграция Pydantic v1 -> v2."""
 
    def leave_FunctionDef(self, original_node, updated_node):
        # @validator -> @field_validator
        if self._has_validator_decorator(updated_node):
            return self._convert_validator(updated_node)
        return updated_node
 
    def leave_ClassDef(self, original_node, updated_node):
        # Config -> model_config = ConfigDict(...)
        if self._has_config_class(updated_node):
            return self._convert_config(updated_node)
        return updated_node
 
    def _convert_validator(self, node):
        """@validator('field') -> @field_validator('field')."""
        new_decorators = []
        for decorator in node.decorators:
            if m.matches(decorator, m.Decorator(decorator=m.Name("validator"))):
                # Заменяем имя декоратора
                new_dec = decorator.with_changes(
                    decorator=cst.Name("field_validator")
                )
                new_decorators.append(new_dec)
            else:
                new_decorators.append(decorator)
 
        # Добавляем @classmethod если нет
        has_classmethod = any(
            m.matches(d, m.Decorator(decorator=m.Name("classmethod")))
            for d in new_decorators
        )
        if not has_classmethod:
            new_decorators.insert(0, cst.Decorator(decorator=cst.Name("classmethod")))
 
        return node.with_changes(decorators=new_decorators)
 
    def _convert_config(self, node):
        """class Config -> model_config = ConfigDict(...)."""
        # Находим Config внутри модели
        config_class = None
        new_body = []
 
        for item in node.body:
            if isinstance(item, cst.ClassDef) and item.name.value == "Config":
                config_class = item
            else:
                new_body.append(item)
 
        if not config_class:
            return node
 
        # Извлекаем параметры из Config
        config_params = self._extract_config_params(config_class)
 
        # Создаём model_config = ConfigDict(...)
        config_dict = cst.Assign(
            targets=[cst.AssignTarget(target=cst.Name("model_config"))],
            value=cst.Call(
                func=cst.Name("ConfigDict"),
                args=[
                    cst.Arg(
                        keyword=cst.Name(k),
                        value=cst.Name(str(v)) if isinstance(v, bool) else cst.SimpleString(f'"{v}"')
                    )
                    for k, v in config_params.items()
                ]
            )
        )
 
        # Вставляем в начало тела класса
        new_body.insert(0, config_dict)
 
        return node.with_changes(body=new_body)
 
# Применяем к файлам
for file_path in all_model_files:
    with open(file_path) as f:
        source = f.read()
 
    tree = cst.parse_module(source)
    transformer = PydanticV2Transformer()
    new_tree = tree.visit(transformer)
 
    with open(file_path, 'w') as f:
        f.write(new_tree.code)

# v1 - работало (но это АНТИПАТТЕРН!)
class Order(BaseModel):
    product_id: int
    quantity: int
 
    @validator('product_id')
    def check_product_exists(cls, v):
        # Запрос в БД внутри валидатора
        product = db.query(Product).get(v)
        if not product:
            raise ValueError(f'Product {v} not found')
        return v
 
# v2 - нужно переписать
class Order(BaseModel):
    product_id: int
    quantity: int
 
    # Валидатор должен быть @classmethod в v2
    @field_validator('product_id')
    @classmethod
    def check_product_exists(cls, v):
        # ⚠️ Проблема: db недоступна в classmethod
        # Решение: убрать I/O из валидатора
        if v <= 0:
            raise ValueError('Product ID must be positive')
        return v
 
    # Проверку существования вынесли в endpoint
    @classmethod
    def create_with_validation(cls, data: dict, db: Session):
        order = cls(**data)
        # Проверяем product после валидации схемы
        product = db.query(Product).get(order.product_id)
        if not product:
            raise HTTPException(404, f'Product {order.product_id} not found')
        return order

# v1
class Event(BaseModel):
    timestamp: datetime
    data: dict
 
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }
 
# v2 - используем field_serializer
class Event(BaseModel):
    timestamp: datetime
    data: dict
 
    @field_serializer('timestamp')
    def serialize_datetime(self, value: datetime) -> str:
        return value.isoformat()

# v1 - работало
for field_name, field in MyModel.__fields__.items():
    print(f"{field_name}: {field.type_}")
 
# v2 - нужно переписать
for field_name, field_info in MyModel.model_fields.items():
    print(f"{field_name}: {field_info.annotation}")

# Прогнал всю тест-сьюту
pytest tests/ -v --cov
 
# Результаты:
Coverage: 87% (было 87% - не изменилось)
Tests passed: 142/145
Tests failed: 3 ← проблемы!

# Модель конфига
class FeatureFlags(BaseModel):
    new_checkout: bool
    beta_features: bool
    debug_mode: bool
 
# В Redis хранились строки "true"/"false"
# v1: приводил "false" -> False ✅
# v2 strict mode: "false" -> True (любая непустая строка) ❌
 
# Результат: все feature flags включились на тестовом стенде

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

# Профилирование памяти
import tracemalloc
 
tracemalloc.start()
 
# До обработки файла
snapshot1 = tracemalloc.take_snapshot()
 
# Обработка 50k заказов
orders = [Order(**row) for row in csv_rows]
process_orders(orders)
 
# После обработки
snapshot2 = tracemalloc.take_snapshot()
 
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
for stat in top_stats[:3]:
    print(stat)
 
# Вывод:
# /app/models/order.py:45: size=842 MiB (+842 MiB), count=50234 (+50234)
# /venv/pydantic/main.py:156: size=234 MiB (+234 MiB), count=150702 (+150702)

# До (с валидацией)
orders = [Order(**row) for row in csv_rows]  # 1.2GB памяти
 
# После (без валидации для trusted данных)
orders = [Order.model_construct(**row) for row in csv_rows]  # 380MB памяти
 
# Валидацию перенесли на этап загрузки CSV (fail fast)
def validate_csv_header(headers):
    expected = set(Order.model_fields.keys())
    if not expected.issubset(headers):
        raise ValueError(f"Missing columns: {expected - set(headers)}")

# kubernetes/api-gateway-canary.yaml
apiVersion: v1
kind: Service
metadata:
  name: api-gateway-canary
spec:
  selector:
    app: api-gateway
    version: v2-pydantic
  ports:
    - port: 8000
 
---
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: api-gateway-traffic-split
spec:
  hosts:
    - api-gateway
  http:
    - match:
        - headers:
            x-canary:
              exact: "true"
      route:
        - destination:
            host: api-gateway-canary
          weight: 100
    - route:
        - destination:
            host: api-gateway-stable
          weight: 95
        - destination:
            host: api-gateway-canary
          weight: 5 # Начал с 5%

# prometheus_metrics.py
from prometheus_client import Histogram, Counter
 
# Метрики для сравнения v1 vs v2
validation_duration = Histogram(
    'pydantic_validation_duration_seconds',
    'Time spent in Pydantic validation',
    ['version', 'model_name']
)
 
validation_errors = Counter(
    'pydantic_validation_errors_total',
    'Total validation errors',
    ['version', 'model_name', 'error_type']
)
 
# В коде
def validate_request(data: dict) -> RequestModel:
    version = "v2" if PYDANTIC_V2_ENABLED else "v1"
 
    with validation_duration.labels(version=version, model_name="RequestModel").time():
        try:
            return RequestModel(**data)
        except ValidationError as e:
            for error in e.errors():
                validation_errors.labels(
                    version=version,
                    model_name="RequestModel",
                    error_type=error['type']
                ).inc()
            raise

# Latency сравнение v1 vs v2
histogram_quantile(0.95,
  rate(pydantic_validation_duration_seconds_bucket[5m])
) by (version)
 
# Error rate v1 vs v2
rate(pydantic_validation_errors_total[5m]) by (version, error_type)

# Было: глобальный strict
class OrderItem(BaseModel):
    model_config = ConfigDict(strict=True)
    id: int
    qty: int
 
# Стало: гибридный подход
class OrderItem(BaseModel):
    model_config = ConfigDict(strict=False)  # По умолчанию flexible
 
    id: int  # Разрешаем "123" -> 123
    qty: int
    price: Decimal = Field(strict=True)  # Но деньги только точные типы!

from my_app.config import settings
 
if settings.PYDANTIC_V2_ENABLED:
    from pydantic.v2 import BaseModel, ValidationError
else:
    from pydantic import BaseModel, ValidationError  # v1

# benchmark_versions.py
import timeit
from typing import List
import pydantic.v1 as pydantic_v1
from pydantic import BaseModel as BaseModelV2
 
# Одинаковые модели на v1 и v2
class UserV1(pydantic_v1.BaseModel):
    id: int
    username: str
    email: str
 
class UserV2(BaseModelV2):
    id: int
    username: str
    email: str
 
# Тестовые данные
data = [
    {"id": i, "username": f"user{i}", "email": f"user{i}@example.com"}
    for i in range(10000)
]
 
# Бенчмарк v1
time_v1 = timeit.timeit(
    lambda: [UserV1(**row) for row in data],
    number=10
) / 10
 
# Бенчмарк v2
time_v2 = timeit.timeit(
    lambda: [UserV2(**row) for row in data],
    number=10
) / 10
 
print(f"v1: {time_v1:.3f}s")
print(f"v2: {time_v2:.3f}s")
print(f"Speedup: {time_v1/time_v2:.1f}x")

# model_dependencies.py
import ast
import networkx as nx
import matplotlib.pyplot as plt
 
def extract_model_dependencies(file_path):
    """Строит граф зависимостей между моделями."""
 
    with open(file_path) as f:
        tree = ast.parse(f.read())
 
    graph = nx.DiGraph()
 
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            # Это Pydantic модель?
            if any(base.id == 'BaseModel' for base in node.bases if isinstance(base, ast.Name)):
                model_name = node.name
                graph.add_node(model_name)
 
                # Ищем зависимости в аннотациях полей
                for item in node.body:
                    if isinstance(item, ast.AnnAssign):
                        annotation = item.annotation
                        # Простой случай: other_field: OtherModel
                        if isinstance(annotation, ast.Name):
                            graph.add_edge(model_name, annotation.id)
                        # List[OtherModel]
                        elif isinstance(annotation, ast.Subscript):
                            if isinstance(annotation.slice, ast.Name):
                                graph.add_edge(model_name, annotation.slice.id)
 
    return graph
 
# Использование
graph = extract_model_dependencies("models/")
 
# Топологическая сортировка для порядка миграции
try:
    migration_order = list(nx.topological_sort(graph))
    print("Миграцировать в порядке:", migration_order)
except nx.NetworkXError:
    print("Обнаружены циклические зависимости!")
    cycles = list(nx.simple_cycles(graph))
    print("Циклы:", cycles)

# test_pydantic_regression.py
import pytest
from decimal import Decimal
 
# Тесты на специфичные баги, которые нашёл
def test_feature_flag_string_bool():
    """Feature flags из Redis должны корректно парситься."""
 
    class Flags(BaseModel):
        enabled: bool
 
    # v1 парсил корректно, v2 нужен strict mode
    assert Flags(enabled="true").enabled is True
    assert Flags(enabled="false").enabled is False  # Важно!
    assert Flags(enabled=True).enabled is True
 
def test_decimal_precision():
    """Деньги должны сохранять точность."""
 
    class Payment(BaseModel):
        amount: Decimal
 
    # Проверяем, что 0.1 + 0.2 = 0.3 точно
    p1 = Payment(amount="0.1")
    p2 = Payment(amount="0.2")
    assert (p1.amount + p2.amount) == Decimal("0.3")
 
def test_model_construct_memory():
    """model_construct не должен создавать лишние объекты."""
 
    import tracemalloc
 
    class Item(BaseModel):
        id: int
        name: str
 
    data = [{"id": i, "name": f"Item {i}"} for i in range(10000)]
 
    tracemalloc.start()
    snapshot1 = tracemalloc.take_snapshot()
 
    items = [Item.model_construct(**row) for row in data]
 
    snapshot2 = tracemalloc.take_snapshot()
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')
 
    # Должно быть меньше 10MB для 10k объектов
    total_size = sum(stat.size for stat in top_stats)
    assert total_size < 10 * 1024 * 1024, f"Memory usage too high: {total_size / 1024 / 1024:.1f}MB"