FlatAdapter для преобразования иерархических структур

Как превратить глубокий 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 FlatAdapterclass OrderItemAdapter(FlatAdapter):    product_id: int    quantity: intclass OrderAdapter(FlatAdapter):    order_id: int    customer_name: str    items: list[OrderItemAdapter]

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

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

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

from typing import Annotatedfrom flat_adapter import Field, FlatAdapterclass OrderItemAdapter(FlatAdapter):    product_id: Annotated[int, Field(source="product.id")]    product_title: Annotated[str, Field(source="product.title")]    quantity: intclass 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: strclass RegionAdapter(FlatAdapter):    region: strclass 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 инструментами.

Теги: Python, ETL, JSON, базы данных, PyPI, data engineering