Граница системы — место, где врут

В прошлом уроке мы построили Order через @dataclass — внутреннюю структуру pipeline. Внутри нашего модуля мы создаём Order сами и доверяем тому, что положили внутрь. Проверки в __post_init__ — больше для подстраховки.

Совсем другая история — на границе системы, там, где данные приходят извне:

• HTTP-ответ от чужого API — может вернуть что угодно, включая null вместо ожидаемого числа.
• Переменная окружения — пользователь поставит PORT=abcd и сломает старт.
• JSON из Kafka — продьюсер изменил схему месяц назад, никого не предупредил.
• CSV из аплоада — внутри попалась дата в формате 15/01/2026 вместо 2026-01-15.

На каждой такой границе нужно проверить, что данные соответствуют ожидаемой структуре. Если нет — отвергнуть с понятной ошибкой, не пуская «грязь» дальше внутрь. Это и есть

Pydantic — это не альтернатива dataclasses. Это дополнение: dataclasses для внутренних DTO, Pydantic для границы. На большом проекте у вас будут и те и другие, и это нормально.

Pydantic v1 и v2

Важный исторический момент: с 2023 года живёт Pydantic v2. Это полностью переписанный с нуля движок на Rust, с другим API.

• v1 → v2 миграция была болезненной, многие методы переименовали.
• v2 быстрее в 5-50 раз в зависимости от сценария.
• В 2026 году писать новый код на v1 нет смысла — это pip install pydantic уже даст вам v2.

В этом курсе — только v2. Если в чужом проекте увидите .parse_obj(), .dict(), .json() — это v1, методы переименованы. Современные:

Минимальная модель

Установим:

uv add pydantic

Самый базовый случай:

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool = True


u = User(id=1, name="Анна", email="[email protected]")
print(u)
# id=1 name='Анна' email='[email protected]' is_active=True

Синтаксис похож на dataclass: class X(BaseModel), поля как аннотации с type hints. Defaults через =. Pydantic читает аннотации и строит валидатор.

Главное отличие: типы проверяются и приводятся в runtime. Если на вход придёт строка "42" где ждали int — Pydantic попробует привести: User(id="42", ...) сработает и u.id будет 42 (int). А User(id="abc", ...) упадёт с ValidationError.

model_validate и model_validate_json

Два основных способа создать модель из «грязных» данных:

# из словаря
raw_dict = {"id": "42", "name": "Анна", "email": "[email protected]"}
u = User.model_validate(raw_dict)

# из JSON-строки (без промежуточного json.loads)
raw_json = '{"id": "42", "name": "Анна", "email": "[email protected]"}'
u = User.model_validate_json(raw_json)

Зачем model_validate_json, когда есть json.loads + model_validate? Скорость и одна точка ошибок. Pydantic v2 умеет парсить JSON на C-уровне сразу в типизированные поля, без промежуточного dict. Для DE с большими payload это заметная разница.

Обратное преобразование:

print(u.model_dump())
# {'id': 42, 'name': 'Анна', 'email': '[email protected]', 'is_active': True}

print(u.model_dump_json())
# {"id":42,"name":"Анна","email":"[email protected]","is_active":true}

print(u.model_dump_json(indent=2))
# {
#   "id": 42,
#   ...
# }

model_dump() — словарь, model_dump_json() — JSON-строка. Это замена asdict() из dataclasses.

ValidationError — главное преимущество

Когда вход не подходит — Pydantic кидает

from pydantic import BaseModel, ValidationError

class User(BaseModel):
    id: int
    name: str
    email: str

bad = {"id": "abc", "name": 123, "email": None}
try:
    User.model_validate(bad)
except ValidationError as e:
    print(e)
    # 3 validation errors for User
    # id
    #   Input should be a valid integer, unable to parse string as an integer
    #   [type=int_parsing, input_value='abc', input_type=str]
    # name
    #   Input should be a valid string [type=string_type, input_value=123, input_type=int]
    # email
    #   Input should be a valid string [type=string_type, input_value=None, input_type=NoneType]

Заметили? Pydantic не валится на первой ошибке — он собирает все. Это критично для DE: за один прогон вы узнаёте, что во всех местах не так с входными данными, не приходится правя по одной ошибке за прогон.

Для логирования используется .errors():

try:
    User.model_validate(bad)
except ValidationError as e:
    for err in e.errors():
        print(err["loc"], err["msg"], err["type"])
    # ('id',) Input should be a valid integer ... int_parsing
    # ('name',) Input should be a valid string string_type
    # ('email',) Input should be a valid string string_type

В production это уходит в structlog с уровнем error (см. урок 06 модуля 4), и в Grafana / Loki видна статистика «какие поля чаще всего ломаются у клиентов».

Constraints — ограничения значений

Кроме типа можно задать ограничения значения через

from pydantic import BaseModel, Field, EmailStr

class User(BaseModel):
    id: int = Field(gt=0)                           # > 0
    name: str = Field(min_length=1, max_length=100) # 1-100 символов
    age: int = Field(ge=0, le=150)                  # 0 <= age <= 150
    email: EmailStr                                  # валидный email
    username: str = Field(pattern=r"^[a-z0-9_]+$")  # regex

Чаще всего нужны:

• gt, ge, lt, le — > / >= / < / <=
• min_length, max_length — для строк и списков
• pattern — regex для строк
• EmailStr — встроенный тип email (требует uv add "pydantic[email]")
• HttpUrl — встроенный валидатор URL

Эти constraints — самое ценное в Pydantic. Бизнес-правила вроде «возраст не отрицательный, email с собакой, username из латиницы» проверяются одной строкой и попадают в .errors() так же, как тип.

Кастомные валидаторы

Если встроенных constraints не хватает — пишем свой через

from pydantic import BaseModel, field_validator

class Order(BaseModel):
    order_id: str
    amount: float
    currency: str

    @field_validator("currency")
    @classmethod
    def currency_is_known(cls, v: str) -> str:
        v = v.upper()
        if v not in ("USD", "EUR", "RUB"):
            raise ValueError(f"unknown currency: {v}")
        return v

    @field_validator("amount")
    @classmethod
    def amount_positive(cls, v: float) -> float:
        if v < 0:
            raise ValueError("amount must be >= 0")
        return v

Несколько моментов:

• @field_validator("field_name") — валидатор для конкретного поля. Можно передать несколько имён сразу.
• @classmethod — обязательно, валидатор это метод класса.
• Функция получает значение, возвращает (возможно изменённое) значение. Если что-то не так — raise ValueError. Pydantic перехватит и подставит в ValidationError.
• Можно нормализовать: v.upper(), v.strip() и т.п.

Для валидации между несколькими полями есть @model_validator(mode="after") — но junior’у он нужен редко.

Composition — вложенные модели

Реальные API-ответы — иерархические. Pydantic умеет это нативно:

from pydantic import BaseModel

class Address(BaseModel):
    city: str
    street: str
    zip: str

class User(BaseModel):
    id: int
    name: str
    address: Address           # вложенная модель
    tags: list[str] = []       # список простых типов

class Team(BaseModel):
    name: str
    members: list[User]        # список вложенных моделей


raw = {
    "name": "platform",
    "members": [
        {
            "id": 1,
            "name": "Анна",
            "address": {"city": "Москва", "street": "Тверская 1", "zip": "101000"},
            "tags": ["lead"],
        },
    ],
}
team = Team.model_validate(raw)
print(team.members[0].address.city)   # 'Москва'

Pydantic рекурсивно валидирует всю иерархию. Если внутри хоть один Address без city — ValidationError с точным путём members.0.address.city.

model_config — настройки модели

В v1 это была class Config:, в v2 — model_config = ConfigDict(...):

from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    model_config = ConfigDict(
        extra="forbid",              # запретить неизвестные поля
        frozen=True,                  # immutable как frozen dataclass
        str_strip_whitespace=True,    # автоматически .strip() у всех str-полей
    )

    id: int
    name: str

Самые полезные для DE опции:

Для production обычно ставят frozen=True, extra="forbid". Это «строгая модель».

DE-кейс: валидация ответа GitHub API

Соберём реалистичный кейс. Дёргаем GitHub API /repos/{owner}/{repo}, парсим ответ.

from pydantic import BaseModel, ConfigDict, Field, HttpUrl

class Owner(BaseModel):
    model_config = ConfigDict(extra="ignore", frozen=True)
    login: str
    id: int
    type: str   # "User" | "Organization"
    html_url: HttpUrl


class Repo(BaseModel):
    model_config = ConfigDict(extra="ignore", frozen=True)

    id: int
    name: str
    full_name: str = Field(pattern=r"^[\w.-]+/[\w.-]+$")
    description: str | None
    owner: Owner
    stargazers_count: int = Field(ge=0)
    forks_count: int = Field(ge=0)
    open_issues_count: int = Field(ge=0)
    language: str | None
    archived: bool
    fork: bool
    html_url: HttpUrl


# использование
import httpx

response = httpx.get("https://api.github.com/repos/astral-sh/uv")
response.raise_for_status()
repo = Repo.model_validate_json(response.content)

print(f"{repo.full_name}: {repo.stargazers_count} stars")
print(f"  owner: {repo.owner.login} ({repo.owner.type})")

Что важно:

• extra="ignore" — GitHub возвращает 50+ полей, нам нужно ~10. Остальные просто игнорируются, без ошибок.
• frozen=True — после парсинга объект нельзя случайно мутировать.
• HttpUrl — проверяет, что это валидный URL. Если придёт "not-a-url" — ошибка.
• Field(ge=0) — счётчики не могут быть отрицательными. Если API сломалось и прислало -1, мы заметим.
• pattern=r"^[\w.-]+/[\w.-]+$" — full_name должен быть в формате owner/repo.
• Вложенный Owner валидируется автоматически.

Если у API случается hiccup и прилетит мусор — мы получим ValidationError с точным указанием поля, не падение в случайном месте через 5 минут с AttributeError: 'NoneType' has no attribute 'stargazers_count'. Это та польза, ради которой Pydantic берут.

Логирование ошибок валидации

В production вы не хотите, чтобы один битый ответ положил весь pipeline. Стандартный паттерн:

import structlog
from pydantic import ValidationError

log = structlog.get_logger(__name__)

def parse_repo(raw_json: bytes) -> Repo | None:
    try:
        return Repo.model_validate_json(raw_json)
    except ValidationError as e:
        log.error(
            "repo_validation_failed",
            errors=e.errors(),
            raw_preview=raw_json[:200].decode("utf-8", errors="replace"),
        )
        return None

В Grafana потом можно сделать дашборд «сколько ответов сегодня прошло валидацию, сколько отвалилось, по каким полям». Это нормальная DE-наблюдаемость над внешним API.

dataclasses vs Pydantic — финальный выбор

Теперь, когда вы видели оба инструмента, правило окончательное:

Правило одной фразой: внутри — dataclass, на границе — Pydantic.

Pydantic «дороже»: добавляет зависимость, валидация занимает время, объект тяжелее dataclass с slots=True. Поэтому внутри pipeline (где данные уже чистые) обычно конвертируют:

def repo_to_record(repo: Repo) -> RepoRecord:
    """Конвертирует pydantic-модель в dataclass для дальнейшего pipeline."""
    return RepoRecord(
        id=repo.id,
        name=repo.full_name,
        stars=repo.stargazers_count,
        ...
    )

Граница есть граница. Один раз отвалидировали — дальше доверяем.

Что должно остаться в голове

1. Pydantic для границы системы: HTTP, env, Kafka, JSON-источники. Всё, чему мы не доверяем.
2. v2 синтаксис: model_validate, model_validate_json, model_dump. Никаких parse_obj / .dict().
3. Constraints через Field(gt=, min_length=, pattern=) заменяют десятки строк if-проверок.
4. ValidationError.errors() даёт все ошибки сразу с путями — идеально для логов.
5. model_config = ConfigDict(extra="forbid", frozen=True) — строгий production-режим.
6. Внутри — dataclass, на границе — Pydantic. Не дублируйте оба для одной задачи.

В следующем уроке — mypy и type checking на CI. Это то, что делает все наши type hints не декоративными, а реально работающими — ловит ошибки до того, как код попал в продакшен.