JSON внутри строки: как проверить вложенный документ Pydantic

Очередь или старый интеграционный шлюз передаёт envelope: обычный JSON с полем payload, внутри которого лежит ещё один JSON, уже строкой. Верхний документ успешно разбирается, но приложение получает текст вместо списка операций. Если объявить payload просто как str, неверная структура внутреннего документа дойдёт до бизнес-логики.

Два этапа проверки на одной границе

Тип Json[T] в Pydantic сначала разбирает JSON-строку, затем проверяет полученное значение как T. В нашем случае T — список операций с положительным количеством. Получается единая ошибка с точным путём до элемента, хотя транспорт содержит два слоя. Не нужно вручную вызывать json.loads, ловить его ошибку и отдельно собирать сообщения об ограничениях каждой строки.

Это подходит именно для JSON, упакованного строкой. Если клиент может передавать нормальный массив, лучше объявить list[Operation] и убрать лишнее кодирование из договора. Json[T] полезен на существующей внешней границе, где формат уже задан. Внутри приложения после проверки работа продолжается с моделями, а не с повторно декодируемой строкой.

Проверим синтаксис, содержимое и повторную загрузку

Понадобится Pydantic v2. Блок можно запустить обычным Python-файлом. Есть три разных отказа: сломанный JSON, объект вместо массива и неверное количество внутри элемента. После успешного разбора посчитаем сумму и проверим, что экспорт подходит для такого же входного формата.

from pydantic import BaseModel, Field, Json, ValidationError

class Operation(BaseModel):
    sku: str
    quantity: int = Field(gt=0)

class Envelope(BaseModel):
    payload: Json[list[Operation]]

for raw in ["{", '{"sku":"A","quantity":2}',
            '[{"sku":"A","quantity":0}]']:
    try:
        Envelope(payload=raw)
    except ValidationError as exc:
        print({"rejected_at": exc.errors()[0]["loc"],
               "code": exc.errors()[0]["type"]})
    else:
        raise AssertionError("invalid payload accepted")

message = Envelope(payload='[{"sku":"A","quantity":2},{"sku":"B","quantity":3}]')
total = sum(op.quantity for op in message.payload)
assert total == 5
normal = message.model_dump()
assert isinstance(normal["payload"], list)
wire = message.model_dump(round_trip=True)
assert isinstance(wire["payload"], str)
restored = Envelope.model_validate(wire)
assert restored == message
print({"quantity": total, "round_trip": restored == message})

Почему обычный dump меняет форму

После валидации поле содержит список объектов. Обычный model_dump отражает это внутреннее представление: в результате получается массив словарей. Но повторный вход Json[T] ждёт JSON-строку. Параметр round_trip=True сохраняет форму, пригодную для повторной валидации этой модели. Это важно для пересылки сообщения в ту же очередь или хранения проверенного envelope.

Если нужно вернуть данные новому HTTP-клиенту, выбирайте формат осознанно. Массив в ответе может быть удобнее двойного JSON, однако тогда входная и выходная модели будут разными. Не определяйте формат API случайно по тому, какой метод сериализации первым оказался под рукой. Зафиксируйте ожидаемые типы полей в тесте, особенно при переходе от старого шлюза к обычному HTTP-договору.

Путь ошибки для количества включает payload, индекс элемента и quantity. Его можно сопоставить исходной операции и отправить интегратору полезную причину отказа. При записи ошибок в журнал избегайте полного payload, если в нём есть личные данные. Для диагностики часто достаточно идентификатора сообщения, индекса и кода ошибки.

Проверка структуры не делает пакет атомарным. Когда проверенные операции записываются в базу, отдельно решите, должен ли отказ одной операции отменить остальные. Валидатор здесь заканчивает работу до записи. Для обычных вложенных HTTP-моделей полезен разбор путей ошибок в строках заказа: там массив приходит без дополнительной строковой упаковки.

Повторная отправка после исправления

Храните идентификатор исходного сообщения отдельно от его содержимого. При отказе интегратор должен понимать, какой пакет исправлять, а сервис — различать новую попытку и другое сообщение с похожими операциями. Сам Json[T] не создаёт дедупликацию: два одинаковых валидных envelope успешно пройдут проверку дважды. Политику повторов задавайте рядом с записью результата обработки.

Если исходная строка подписана отправителем, подпись проверяют по исходным байтам до преобразования. Повторная сериализация может изменить пробелы и порядок представления, даже когда смысл JSON сохранился. round_trip обеспечивает пригодность для модели, но не обещает сохранение байтов исходного транспорта. Поэтому проверенный объект и исходное подписанное сообщение выполняют разные задачи и не должны подменять друг друга.

Продолжить на практике

Следующий полезный шаг — отделить проверку входа от обработки операций и покрыть обе части тестами. В продвинутом Python есть pytest, собственные исключения и Repository; эти материалы помогут оформить такую границу вокруг интеграции. Откройте программу продвинутого Python.