Как превратить глубокий JSON в строки для БД с помощью pydantic-подобной иерархии Python-классов

Проблема

На одном из предыдущих проектов у меня возникла типичная ETL-задача: из глубоко вложенного JSON нужно было извлечь данные и разложить их по нескольким таблицам базы данных.

Упрощённый пример входных данных выглядел так:

{
  "order_id": "1001",
  "customer": {
    "id": "42",
    "name": "Alice"
  },
  "items": [
    {
      "product": {
        "id": "10",
        "title": "Keyboard"
      },
      "quantity": "2"
    },
    {
      "product": {
        "id": "20",
        "title": "Mouse"
      },
      "quantity": "1"
    }
  ]
}

На выходе хотелось получить плоские строки:

[
    {
        "order_id": 1001,
        "customer_name": "Alice",
        "product_id": 10,
        "quantity": 2,
    },
    {
        "order_id": 1001,
        "customer_name": "Alice",
        "product_id": 20,
        "quantity": 1,
    },
]

Такие строки уже удобно вставлять в staging-таблицу или передавать дальше в ETL-пайплайне.

Почему не написать один рекурсивный парсер

Первую версию такого решения обычно легко написать за вечер. Сложность начинается позже:

• у каждого поля появляется свой путь в JSON;

• значения нужно приводить к int, float, bool, date;

• часть полей необязательная;

• вложенные списки должны размножать строки;

• два разных вложенных объекта могут породить одинаковое имя колонки;

• ошибки нужно связывать с конкретным полем;

• глубокая рекурсия не должна незаметно создать миллионы строк.

В итоге появляется код с большим количеством if, обращения индексу и ручных преобразований. Хотелось описывать структуру декларативно, примерно в стиле pydantic.BaseModel, но не создавать ещё одну модель валидации и не привязывать решение к ORM.

Так появилась идея FlatAdapter.

Иерархия adapters

Установка минимальна:

pip install flat-adapter

Описание структуры можно сделать через обычные Python-классы:

from flat_adapter import FlatAdapter


class OrderItemAdapter(FlatAdapter):
    product_id: int
    quantity: int


class OrderAdapter(FlatAdapter):
    order_id: int
    customer_name: str
    items: list[OrderItemAdapter]

Аннотации задают типы и одновременно описывают будущие колонки. Вложенный FlatAdapter обозначает mapping, а list[FlatAdapter] — повторяющуюся структуру, которая должна развернуться в несколько строк.

Пути и преобразования

Для нестандартного пути используется Field. В проектах с mypy --strict рекомендуется записывать metadata через typing.Annotated:

from typing import Annotated

from flat_adapter import Field, FlatAdapter


class OrderItemAdapter(FlatAdapter):
    product_id: Annotated[int, Field(source="product.id")]
    product_title: Annotated[str, Field(source="product.title")]
    quantity: int


class OrderAdapter(FlatAdapter):
    order_id: int
    customer_name: Annotated[str, Field(source="customer.name")]
    items: list[OrderItemAdapter]

Исходный payload для этого adapter:

payload = {
    "order_id": "1001",
    "customer": {"name": "Alice"},
    "items": [
        {
            "product": {"id": "10", "title": "Keyboard"},
            "quantity": "2",
        },
        {
            "product": {"id": "20", "title": "Mouse"},
            "quantity": "1",
        },
    ],
}

Теперь адаптация входного mapping выглядит так:

rows = OrderAdapter.adapt(payload)

Строковые значения "1001", "10" и "2" будут приведены к указанным аннотациям.

У поля можно задать значение по умолчанию и функцию подготовки:

class CustomerAdapter(FlatAdapter):
    name: Annotated[
        str,
        Field(
            source="customer.name",
            default="Unknown",
            prepare_data_func=lambda value: str(value).strip(),
        ),
    ]

Legacy-вариант name: str = Field(...) также поддерживается во время выполнения, но Annotated лучше согласуется со строгой статической типизацией.

Как разложить данные по нескольким таблицам

FlatAdapter не знает о базе данных и намеренно не содержит repository или ORM-слой. Это позволяет отдельно описать строки для каждой целевой таблицы:

class OrderTableAdapter(FlatAdapter):
    order_id: int
    customer_id: Annotated[int, Field(source="customer.id")]
    customer_name: Annotated[str, Field(source="customer.name")]


class OrderItemsTableAdapter(FlatAdapter):
    order_id: int
    items: list[OrderItemAdapter]

Здесь OrderItemsTableAdapter — верхнеуровневый adapter для таблицы items. Поэтому отдельный цикл по payload["items"] не нужен:

order_rows = OrderTableAdapter.adapt(payload)
item_rows = OrderItemsTableAdapter.adapt(payload)

После этого order_row и item_rows можно передать в отдельный слой загрузки в БД. Такое разделение важно: адаптер преобразует данные, но не решает за приложение вопросы транзакций, retry и bulk insert.

Что происходит со списками

Если в adapter есть несколько независимых списков, результатом становится декартово произведение:

class TagAdapter(FlatAdapter):
    tag: str


class RegionAdapter(FlatAdapter):
    region: str


class ProductAdapter(FlatAdapter):
    tags: list[TagAdapter]
    regions: list[RegionAdapter]

Пример исходного payload:

payload = {
    "tags": [{"tag": "python"}, {"tag": "etl"}],
    "regions": [
        {"region": "eu"},
        {"region": "us"},
        {"region": "apac"},
    ],
}

Два тега и три региона дадут шесть строк. Порядок входных списков сохраняется, а пустой или None список оставляет родительскую строку с None в дочерних колонках.

Такое поведение удобно для ETL, но у него есть очевидный риск. Если длины списков равны L1, L2, …, число строк может расти как:

R = product(max(1, len(Li)))

Поэтому для внешних данных стоит задавать защитный лимит:

rows = OrderAdapter.adapt(payload, max_rows=10_000)

При превышении лимита адаптер выбрасывает RowLimitExceeded, не возвращая частичный результат.

Что с глубокой вложенностью и производительностью

Семь-десять уровней вложенности сами по себе не выглядят проблемой для рекурсивного Python-обхода. Главный фактор — не глубина, а число комбинаций и колонок в результирующих строках.

В библиотеке есть отдельный локальный benchmark:

uv run python benchmarks/flatten_benchmark.py

Он измеряет:

• смешанную структуру глубиной 10 уровней;

• список из 1000 элементов;

• произведение 10 × 10 × 10;

• адаптер с 20 плоскими колонками.

Метод adapt() возвращает list[dict[str, object]], поэтому строки полностью материализуются в памяти. Для больших потоков данных можно использовать iter_adapt(): он выдаёт строки лениво и совместим с max_rows, что позволяет обрабатывать результат по одной строке и не держать весь набор в памяти.

Ошибки вместо тихой порчи данных

Адаптер не должен молча перезаписывать колонку или возвращать частичный результат. Для этого выделены типизированные ошибки:

• MissingFieldError — обязательное поле отсутствует;

• ConversionError — значение не удалось привести к типу;

• InvalidShapeError — вложенная структура имеет неверную форму;

• DuplicateFieldError — два nested adapter породили одну колонку;

• RowLimitExceeded — expansion превысил заданный лимит.

Что библиотека не делает

Это не замена Pydantic, ORM и ETL-платформе:

• нет HTTP API;

• нет работы с БД;

• нет схем миграций;

• нет валидации бизнес-правил;

• нет поддержки dataclass и Pydantic-моделей в первой версии.

Задача библиотеки уже и проще: предсказуемо превратить nested mapping в строки, которые можно передать следующему слою.

Итоги

Идея иерархии adapters оказалась удобной границей между форматом входных данных и persistence-слоем:

1. структура JSON видна в коде;

2. типы и преобразования находятся рядом с полями;

3. вложенные списки превращаются в строки детерминированно;

4. опасное декартово расширение можно ограничить;

5. адаптер не связан с конкретной БД.

Проект опубликован на PyPI:

• PyPI

• GitHub

• English documentation

Если у вас была похожая задача — интересно сравнить этот подход с вашими решениями: ручными flatten-функциями, Pydantic-моделями или полноценными ETL инструментами.