Как превратить глубокий 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-слоем:
структура JSON видна в коде;
типы и преобразования находятся рядом с полями;
вложенные списки превращаются в строки детерминированно;
опасное декартово расширение можно ограничить;
адаптер не связан с конкретной БД.
Проект опубликован на PyPI:
Если у вас была похожая задача — интересно сравнить этот подход с вашими решениями: ручными flatten-функциями, Pydantic-моделями или полноценными ETL инструментами.
