Оглавление

1. Введение

2. Ленивая стратегия

3. Selectin

4. Joinload

5. Сырой рекурсивный CTE-запрос

6. CTE-запрос core SQLAlchemy

7. Сравнение производительности

Введение

Когда возникает необходимость работать с иерархической структурой данных, кажется, что решение давно найдено, ведь подобные задачи уже неоднократно решались. Возможно, даже выбран инструмент, например, Python и SQLAlchemy. Однако, углубляясь в задачу, понимаешь, что вариантов множество, даже в вопросе, как извлечь данные из базы: использовать стратегию selectin для загрузки родительских элементов? Или, возможно, стоит применить joinload? А может, лучше воспользоваться CTE‑запросами, которые рекомендуются в 90% статей? Но насколько CTE действительно эффективно по сравнению с другими методами? Более того, большинство примеров рассматривают ситуации в идеальных условиях, далёких от реальных проектов.

В этой статье я рассмотрю основные способы получения иерархической структуры из БД на примере реального многослойного приложения с использованием SQLAlchemy 2.0. Как обычно, есть важные нюансы, о которых редко упоминают, хотя они весьма любопытны. В завершение проведу сравнение производительности всех описанных подходов.

Сначала определимся с доменными именами. Допустим, бизнесу понадобилось структурировать проекты в компании и разложить их по полочкам, а в этой компании «Рога и Копыта» все конечности располагались в одной куче. Для структурирования проектов будем использовать директории, но в компании мы дали своё доменное имя — пространство. В пространстве могут располагаться проекты и подпространства. Да-да, задача донельзя приземленная.

Теперь рассмотрим результат, который я хочу получить из API, запрашивая пространство, — красивую вложенную структуру целевого пространства и всех его родителей:

{
    "result": {
        "id": 13,
        "name": "Cool Space",
        "parent": {
            "id": 10,
            "name": "Intermediate space",
            "parent": {
                "id": 7,
                "name": "Root space",
                "parent": null
            }
        }
    },
    "errors": []
}

Пробежимся по слоям приложения и не будем акцентировать на этом внимание. Определим доменную сущность:

from __future__ import annotations

from pydantic import Field

from app.entities.base import BaseEntity


class SpaceEntity(BaseEntity):
    id: int
    name: str
    parent: SpaceEntity | None = Field(default=None)

Ручка запроса пространства по его id:

@router.get("/spaces/{space_id}", response_model=SpaceResponse)
@inject
async def create_space(
    space_id: int, space_service: SpaceService = Depends(Provide[Container.services.space])
):
    space = await space_service.get_three_by_id(space_id)
    return {"result": space}

И сервис приложения. На этом уровне работаем с паттерном UnitOfWork, управляя транзакциями и репозиториями. В данном случае мы вызываем репозиторий и пытаемся достать сущность пространства из базы данных по id.

class SpaceService:

    def __init__(self, uow: IUnitOfWorkBase) -> None:
        self._uow = uow

    async def get_three_by_id(self, space_id: int) -> SpaceEntity:
        async with self._uow as uow:
            space = await uow.spaces.get_by_id(
                id=space_id
            )

        return space

На этом мы останавливаемся, ведь в зависимости от выбранного пути получения иерархической структуры будет меняться реализация метода репозитория get_by_id.

Ленивая стратегия

Полный код проекта для ленивой стратегии можно посмотреть тут. Помимо метода репозитория нужно объявить и модель базы данных:

class SpaceModel(Base):
    __tablename__ = "spaces"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at: Mapped[datetime] = mapped_column(DateTime, onupdate=datetime.datetime.now)
    parent_id: Mapped[int] = mapped_column(Integer, ForeignKey("spaces.id"))

    parent: Mapped[Optional['SpaceModel']] = relationship(remote_side=[id])

    def __repr__(self):
        return f"<SpaceModel(id={self.id}, " \
               f"name=\"{self.name}\", " \
               f"parent_id=\"{self.parent_id}\", " \
               f"created_at=\"{self.created_at}\", " \
               f"updated_at=\"{self.updated_at}\">"

Пространство связывается с родительским через parent_id, где внешний ключ (foreign key) ссылается на столбец id этой же таблички (самореференсная ссылка). И также объявляется атрибут parent, который представляет собой отношение (relationship). К этому атрибуту мы будем в дальнейшем обращаться, чтобы получить родительское пространство.

Метод репозитория запроса сущности по id :

async def get_by_id(self, space_id: int) -> SpaceEntity | None:
    result = (await self._session.scalars(
        select(SpaceModel).where(SpaceModel.id == space_id)
    )).unique().one_or_none()

    return await map_to_entity(result)

После выполнения запроса к БД на выходе мы получим только пространство с указанным id. Рекурсия же достигается ленивой подгрузкой (lazy‑стратегия) родительского пространства обращением к атрибуту parent. Запрашивать будем до тех пор, пока не достигнем необходимого условия остановки, в нашем случае — пока не получим корневое пространство (parent = None).

В синхронном режиме SQLAlchemy, чтобы сделать дополнительный запрос к базе за родительским пространством, необходимо просто вызвать атрибут parent.

model.parent

В асинхронном режиме при обращении к parent будет ошибка:

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/20/xd2s)

Чтобы lazy-стратегия заработала в асинхронном режиме, SQLAlchemy нужно в базовый класс Base моделей добавить родителя AsyncAttrs, и уже после этого можно обращаться к parent, но только через property-атрибут awaitable_attrs.

await model.awaitable_attrs.parent

Функция маппинга модели БД SpaceModel в доменную модель SpaceEntity:

async def map_to_entity(model: SpaceModel | None) -> SpaceEntity | None:
    if not model:
        return None

    parent = await model.awaitable_attrs.parent

    return SpaceEntity(
        id=model.id,
        name=model.name,
        parent=await map_to_entity(parent)
    )

Здесь и реализовано рекурсивное получение данных— асинхронно ходим в базу за родительским пространством, пока не упрёмся в корневое пространство, а уже после этого на обратном пути маппим модель БД SpaceModel в доменную сущность SpaceEntity.

В этом рекурсивном способе получения данных нет защиты от зацикливания! Если, скажем, существует пространство с id=1, а его родительское пространство parent_id=1, то мы, конечно, попадём в ловушку.

Selectin

Переходим к более интересной стратегии подгрузки данных — selectin. Весь проект и реализацию стратегии selectin можно посмотреть в данной ветке репозитория.

Рассмотрим модель БД, она не сильно поменялась:

DEPTH = 5

class SpaceModel(Base):
    __tablename__ = "spaces"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at: Mapped[datetime] = mapped_column(DateTime, onupdate=datetime.datetime.now)
    parent_id: Mapped[int] = mapped_column(Integer, ForeignKey("spaces.id"))

    parent: Mapped[Optional['SpaceModel']] = relationship(remote_side=[id], lazy="selectin", join_depth=DEPTH)

    def __repr__(self):
        return f"<SpaceModel(id={self.id}, " \
               f"name=\"{self.name}\", " \
               f"parent_id=\"{self.parent_id}\", " \
               f"created_at=\"{self.created_at}\", " \
               f"updated_at=\"{self.updated_at}\">"

В отношении (relationship) указываем стратегию подгрузки данных lazy="selectin", и появляется новый параметр join_depth. Он отвечает за глубину подгружаемых данных, или, проще говоря, за количество дополнительных запросов к БД. «Алхимия» берет id родительского пространства и подгружает обычным select'ом, и так DEPTH раз. Например, вот так в консоли при включенном отладочного режиме в «Алхимии» будет выглядеть результат выполнения запроса:

result = (await self._session.scalars(
    select(SpaceModel).where(SpaceModel.id == id)
)).unique().one_or_none()

-----------------------------------------------

INFO sqlalchemy.engine.Engine BEGIN (implicit)
FROM spaces 
WHERE spaces.id IN ($1::INTEGER)
INFO sqlalchemy.engine.Engine [cached since 51.12s ago] (13,)
INFO sqlalchemy.engine.Engine SELECT spaces.id AS spaces_id, spaces.name AS spaces_name, spaces.created_at AS spaces_created_at, spaces.updated_at AS spaces_updated_at, spaces.parent_id AS spaces_parent_id 
FROM spaces 
WHERE spaces.id IN ($1::INTEGER)
INFO sqlalchemy.engine.Engine [cached since 51.12s ago] (10,)
INFO sqlalchemy.engine.Engine SELECT spaces.id AS spaces_id, spaces.name AS spaces_name, spaces.created_at AS spaces_created_at, spaces.updated_at AS spaces_updated_at, spaces.parent_id AS spaces_parent_id 
FROM spaces 
WHERE spaces.id IN ($1::INTEGER)
INFO sqlalchemy.engine.Engine [generated in 0.00040s] (7,)
INFO sqlalchemy.engine.Engine SELECT spaces.id AS spaces_id, spaces.name AS spaces_name, spaces.created_at AS spaces_created_at, spaces.updated_at AS spaces_updated_at, spaces.parent_id AS spaces_parent_id 
FROM spaces 
WHERE spaces.id IN ($1::INTEGER)
INFO sqlalchemy.engine.Engine [generated in 0.00039s] (4,)
INFO sqlalchemy.engine.Engine SELECT spaces.id AS spaces_id, spaces.name AS spaces_name, spaces.created_at AS spaces_created_at, spaces.updated_at AS spaces_updated_at, spaces.parent_id AS spaces_parent_id 
FROM spaces 
WHERE spaces.id = $1::INTEGER
INFO sqlalchemy.engine.Engine [generated in 0.00094s] (1,)
INFO sqlalchemy.engine.Engine ROLLBACK

После выполнения запроса «Алхимия» сама возьмёт подгруженные пространства и правильно организует вложенную структуру. Но здесь запрашивается только небольшое количество подпространств, а нам нужно получить всех родителей. Полный метод репозитория будет выглядеть так:

async def get_by_id(self, id: int) -> SpaceEntity | None:
    result = (await self._session.scalars(
        select(SpaceModel).where(SpaceModel.id == id)
    )).unique().one_or_none()

    return await map_to_entity(result, DEPTH)

Вся рекурсивная работа, как и в предыдущем разделе с ленивой подгрузкой, будет выполняться в функции маппинга map_to_entity:

async def map_to_entity(model: SpaceModel | None, depth: int) -> SpaceEntity | None:
    if not model:
        return None

    if depth == 0:
        parent_model = await model.awaitable_attrs.parent
        parent_entity = await map_to_entity(parent_model, DEPTH - 1)
    else:
        parent_entity = await map_to_entity(model.parent, depth - 1)

    return SpaceEntity(
        id=model.id,
        name=model.name,
        parent=parent_entity
    )

Функция несколько усложнилась, теперь мы спускаемся на глубину, равную DEPTH (в нашем случае — 5). потому что пять подпространств уже загружены. Теперь на шестом обращении к атрибуту parent мы получим уже знакомое исключение:

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/20/xd2s)

Поэтому нужно опять обращаться к parent через свойство awaitable_attrs, чтобы «Алхимия» асинхронно сходила в базу и собрала пять подпространств. Потом мы обнуляем счётчик и опять спускаемся в глубину на пять пространств, чтобы запросить новую партию. И так до тех пор, пока не упрёмся в корневое пространство.

По сути, это тот же lazy, только мы получаем пачку данных (n=depth). Несложно догадаться, что количество запросов будет такое же, как и при lazy‑стратегии.

В этом рекурсивном способе получения данных нет защиты от зацикливания! Если, скажем, существует пространство с id=1, а его родительское пространство parent_id=1, то мы, конечно, попадём в ловушку.

Если не нужно получать все дерево вплоть до корневого элемента, но и заранее не хочется задавать глубину вложенности в классе модели SpaceModel есть альтернативное решение:

async def get_by_id(self, id: Union[int, UUID]) -> SpaceEntity | None:
    parent_alias = aliased(SpaceModel)

    result = (await self._session.scalars(
        select(SpaceModel).options(
            selectinload(SpaceModel.parent, recursion_depth=DEPTH),
        ).where(SpaceModel.id == id).join(SpaceModel.parent.of_type(parent_alias), full=True)
    )).unique().one_or_none()

    return await map_to_entity(result, DEPTH)


async def map_to_entity(model: SpaceModel | None, depth: int) -> SpaceEntity | None:
    if not model:
        return None

    if depth == 0:
        return None

    return SpaceEntity(
        id=model.id,
        name=model.name,
        parent=await map_to_entity(model.parent, depth - 1)
    )

Joinload

Наконец-то поговорим о чём‑то более производительном. Весь проект и реализацию стратегии joinload можно посмотреть в данной ветке репозитория. В модели БД изменение совсем небольшое, нужно поменять стратегию подгрузки —lazy="joined":

DEPTH = 5

class SpaceModel(Base):
    __tablename__ = "spaces"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at: Mapped[datetime] = mapped_column(DateTime, onupdate=datetime.datetime.now)
    parent_id: Mapped[int] = mapped_column(Integer, ForeignKey("spaces.id"))

    parent: Mapped[Optional['SpaceModel']] = relationship(remote_side=[id], lazy="joined", join_depth=DEPTH)

    def __repr__(self):
        return f"<SpaceModel(id={self.id}, " \
               f"name=\"{self.name}\", " \
               f"parent_id=\"{self.parent_id}\", " \
               f"created_at=\"{self.created_at}\", " \
               f"updated_at=\"{self.updated_at}\">"

Метод репозитория не изменился:

async def get_by_id(self, id: Union[int, UUID]) -> SpaceEntity | None:
    result = (await self._session.scalars(
        select(SpaceModel).where(SpaceModel.id == id)
    )).unique().one_or_none()

    return await map_to_entity(result, DEPTH)

Но изменились запросы к БД. Посмотрим в консоль на результат:

result = (await self._session.scalars(
    select(SpaceModel).where(SpaceModel.id == id)
)).unique().one_or_none()

---------------------------------------------------

INFO sqlalchemy.engine.Engine 
SELECT spaces.id, spaces.name, spaces.created_at, spaces.updated_at, spaces.parent_id, 
       spaces_1.id AS id_1, spaces_1.name AS name_1, spaces_1.created_at AS created_at_1, spaces_1.updated_at AS updated_at_1, spaces_1.parent_id AS parent_id_1, 
       spaces_2.id AS id_2, spaces_2.name AS name_2, spaces_2.created_at AS created_at_2, spaces_2.updated_at AS updated_at_2, spaces_2.parent_id AS parent_id_2, 
       spaces_3.id AS id_3, spaces_3.name AS name_3, spaces_3.created_at AS created_at_3, spaces_3.updated_at AS updated_at_3, spaces_3.parent_id AS parent_id_3, 
       spaces_4.id AS id_4, spaces_4.name AS name_4, spaces_4.created_at AS created_at_4, spaces_4.updated_at AS updated_at_4, spaces_4.parent_id AS parent_id_4, 
       spaces_5.id AS id_5, spaces_5.name AS name_5, spaces_5.created_at AS created_at_5, spaces_5.updated_at AS updated_at_5, spaces_5.parent_id AS parent_id_5 
FROM spaces 
       LEFT OUTER JOIN spaces AS spaces_5 ON spaces_5.id = spaces.parent_id 
       LEFT OUTER JOIN spaces AS spaces_4 ON spaces_4.id = spaces_5.parent_id 
       LEFT OUTER JOIN spaces AS spaces_3 ON spaces_3.id = spaces_4.parent_id 
       LEFT OUTER JOIN spaces AS spaces_2 ON spaces_2.id = spaces_3.parent_id 
       LEFT OUTER JOIN spaces AS spaces_1 ON spaces_1.id = spaces_2.parent_id 
WHERE spaces.id = $1::INTEGER
INFO sqlalchemy.engine.Engine [generated in 0.00152s] (13,)

Как видите, «Алхимия» собрала пять JOIN в один запрос. Очевидно, что этот запрос будет более производительный, чем пять дополнительных в selectin‑стратегии. После выполнения запроса «Алхимия» всё так же аккуратно соберёт данные, создаст модели и организует вложенность.

Функция маппинга из предыдущей главы тоже не изменилась:

async def map_to_entity(model: SpaceModel | None, depth: int) -> SpaceEntity | None:
    if not model:
        return None

    if depth == 0:
        parent_model = await model.awaitable_attrs.parent
        parent_entity = await map_to_entity(parent_model, DEPTH)
    else:
        parent_entity = await map_to_entity(model.parent, depth - 1)

    return SpaceEntity(
        id=model.id,
        name=model.name,
        parent=parent_entity
    )

При первом входе в функцию мы уже имеем модель SpaceModel с подгруженными пятью вложенными пространствами, поэтому мы просто спускаемся вниз по родителям, уменьшая счётчик с 5 до 0. Как только счетчик depth становится равен нулю, нужно заново подгрузить данные. Как уже сказано выше, подгрузка будет осуществляться через свойство awaitable_attrs. «Алхимия» снова асинхронно сходит в базу и за один запрос (с пятью JOIN) заберёт пространства. Обнуляем счетчик и опять спускаемся вниз. И так до тех пор, пока не уткнёмся в родительское пространство (parent=None).

В этом рекурсивном способе получения данных нет защиты от зацикливания!

Есть альтернативное решение:

    async def get_by_id(self, id: Union[int, UUID]) -> SpaceEntity | None:
        parent_alias = aliased(SpaceModel)
    
        result = (await self._session.scalars(
            select(SpaceModel).options(
                joinedload(SpaceModel.parent).joinedload(SpaceModel.parent).joinedload(SpaceModel.parent),
            ).where(SpaceModel.id == id).join(SpaceModel.parent.of_type(parent_alias), full=True)
        )).unique().one_or_none()
    
        return await map_to_entity(result, 3)


async def map_to_entity(model: SpaceModel | None, depth: int) -> SpaceEntity | None:
    if not model:
        return None

    if depth == 0:
        return None

    return SpaceEntity(
        id=model.id,
        name=model.name,
        parent=await map_to_entity(model.parent, depth - 1)
    )

Сырой рекурсивный CTE-запрос

Если вы будете искать, как реализовать рекурсивный запрос, то 90% статей будет на тему CTE‑запроса. Ну, раз советуют, то надо попробовать. Не будем сейчас останавливаться на том, как это работает. Предлагаю лишь почитать отличную статью на эту тему. Весь проект и реализацию рекурсивного запроса сырым CTE можно посмотреть в данной ветке репозитория.

Начнём, как обычно, с модели БД, которая немного упростилась. Теперь нам не нужно создавать relationship, собирать и маппить данные будем своими силами. Модель выглядит теперь так:

class SpaceModel(Base):
    __tablename__ = "spaces"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at: Mapped[datetime] = mapped_column(DateTime, onupdate=datetime.datetime.now)
    parent_id: Mapped[int] = mapped_column(Integer, ForeignKey("spaces.id"))

    def __repr__(self):
        return f"<SpaceModel(id={self.id}, " \
               f"name=\"{self.name}\", " \
               f"parent_id=\"{self.parent_id}\", " \
               f"created_at=\"{self.created_at}\", " \
               f"updated_at=\"{self.updated_at}\">"

Готовый чистый SQL‑запрос для получения пространства и всех его родителей:

WITH RECURSIVE parent_table(space_id, space_name, parent_id) AS (
	SELECT id as space_id, name as space_name, parent_id
	FROM spaces WHERE id = {id}
	UNION
		SELECT s.id as space_id, s.name as space_name, s.parent_id
		FROM spaces as s
	INNER JOIN parent_table p ON p.space_id = s.parent_id
) 
SELECT * FROM parent_table

Но давайте его усложним и добавим защиту от зацикливания:

WITH RECURSIVE parent_table(id, name, parent_id, ids, cycle) AS (
SELECT id, name, parent_id, ARRAY[id], false
FROM spaces WHERE id = :val
UNION
    SELECT s.id, s.name, s.parent_id, s.id || p.ids, s.id = ANY(p.ids)
    FROM spaces as s
INNER JOIN parent_table p ON p.parent_id = s.id
WHERE not cycle
)
SELECT * FROM parent_table

Теперь перенесём этот запрос прямо в метод репозитория:

async def get_by_id(self, id: int) -> SpaceEntity | None:
    result = (await self._session.execute(
        text("""
            WITH RECURSIVE parent_table(id, name, parent_id, ids, cycle) AS (
            SELECT id, name, parent_id, ARRAY[id], false
            FROM spaces WHERE id = :val
            UNION
                SELECT s.id, s.name, s.parent_id, s.id || p.ids, s.id = ANY(p.ids)
                FROM spaces as s
            INNER JOIN parent_table p ON p.parent_id = s.id
            WHERE not cycle
            )
            SELECT * FROM parent_table
        """).bindparams(val=id),
    )).fetchall()

    spaces_dict = {elem.id: elem for elem in result}

    return to_three(id, spaces_dict) if result else None

В отличие от предыдущих глав, теперь функция маппинга, отвечает только за создание SpaceEntity, больше никаких дополнительных запросов к базе.

Важный момент, который нигде не подсвечивается: результатом выполнения запроса будет плоский список кортежей, где каждый кортеж состоит из значений столбцов, которые объявлены в parent_table. Далее нам самим придётся маппить кортежи в доменную сущность и создавать необходимую вложенность.

Перед вызовом функции to_three выполним небольшую оптимизацию: создадим словарь, где ключом будет id пространства, а значением — модель БД SpaceModel. В самой функции пройдёмся по элементам этого словаря и создадим словарь, но уже со значением доменной сущности и заодно будем проставлять каждой сущности ссылку на родительский объект. В конце достанем из словаря нужное пространство и вернём его.

def to_three(space_id: int, models: dict[int, SpaceModel]) -> SpaceEntity:
    entities = {
        int(model.id): SpaceEntity(id=model.id, name=model.name)
        for model in models.values()
    }

    for _, v in models.items():

        if v.parent_id not in entities:
            continue

        parent_space = entities.get(v.parent_id)
        if not parent_space:
            raise Exception("Parent space is not found")

        space = entities.get(v.id)
        if not space:
            raise Exception("Space is not found")

        space.parent = parent_space

    return entities[space_id]

Первый кандидат на звание самого производительного решения. Верно?

CTE-запрос core SQLAlchemy

И в заключение попробуем преобразовать сырой CTE в запрос в стиле core SQLAlchemy. Посмотреть весь проект и реализацию рекурсии с помощью CTE можно посмотреть в данной ветке репозитория. И как всегда, начнём с модели БД:

class SpaceModel(Base):
    __tablename__ = "spaces"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at: Mapped[datetime] = mapped_column(DateTime, onupdate=datetime.datetime.now)
    parent_id: Mapped[int] = mapped_column(Integer, ForeignKey("spaces.id"))

    parent: Mapped[Optional['SpaceModel']] = relationship(remote_side=[id])

    def __repr__(self):
        return f"<SpaceModel(id={self.id}, " \
               f"name=\"{self.name}\", " \
               f"parent_id=\"{self.parent_id}\", " \
               f"created_at=\"{self.created_at}\", " \
               f"updated_at=\"{self.updated_at}\">"

Здесь мы вернули атрибут parent. Зачем — расскажу ниже. Теперь как будет выглядеть SQLAlchemy statement:

alias = aliased(SpaceModel, name='s')
parent_table = select(SpaceModel).where(SpaceModel.id == id).cte("parent_table", recursive=True)

q = parent_table.union(
    select(alias).join(
        parent_table, parent_table.c.parent_id == alias.id
    )
)

result = (await self._session.scalars(
    select(aliased(SpaceModel, alias=q))
)).unique().fetchall()

Всё бы ничего, но мы забыли про защиту от зацикливания. Полный core statement c выполнением запроса в методе репозитория:

async def get_by_id(self, id: int) -> SpaceEntity | None:
  """
  Чистый sql запрос:
      WITH RECURSIVE parent_table(id, name, parent_id, ids, cycle) AS (
      SELECT id, name, parent_id, ARRAY[id], false
      FROM spaces WHERE id = :val
      UNION
          SELECT s.id, s.name, s.parent_id, s.id || p.ids, s.id = ANY(p.ids)
          FROM spaces as s
      INNER JOIN parent_table p ON p.parent_id = s.id
      WHERE not cycle
      )
      SELECT * FROM parent_table
  :param space_id:
  :return:
  """

  parent_table = (
      select(
          SpaceModel.id,
          SpaceModel.name,
          SpaceModel.parent_id,
          SpaceModel.created_at,
          SpaceModel.updated_at,
          (array([column("id")])).label("ids"),
          literal(False).label('cycle')
      )
      .where(SpaceModel.id == id)
      .cte("parent_table", recursive=True)
  )
  space_alias = aliased(SpaceModel, name='s')

  query = parent_table.union(
      select(
          space_alias.id,
          space_alias.name,
          space_alias.parent_id,
          space_alias.created_at,
          space_alias.updated_at,
          (column('ids').op('||')(space_alias.id)).label("ids"),
          parent_table.c.ids.any(space_alias.id).label("cycle")
      )
      .join(parent_table, parent_table.c.parent_id == space_alias.id)
      .where(parent_table.c.cycle == False)
  )

  result = (await self._session.scalars(
      select(aliased(SpaceModel, alias=query))
  )).unique().fetchall()

  return map_to_entity(
      result[0] if result else None
  )  # noqa

Выглядит, конечно, страшно. Но большое преимущество этой реализации в том, что «Алхимия» не только создаст модели пространств SpaceModel, но и сама организует вложенность. Нам не придётся это делать собственноручно.

Наконец, простая функция преобразования из модели базы данных SpaceModel в доменную сущность SpaceEntity выглядит следующим образом:

def map_to_entity(model: SpaceModel | None) -> SpaceEntity | None:
    if not model:
        return None

    return SpaceEntity(
        id=model.id,
        name=model.name,
        parent=map_to_entity(model.parent)
    )

Для этого примера реализована защита от зацикливания!

Теперь перейдём к самому интересному: сравнению производительности!

Сравнение производительности

Давайте прямо сравним все рассмотренные методы получения иерархической структуры. Будем измерять производительность не отдельного запроса, а всего метода репозитория get_by_id, поскольку во многих представленных решениях присутствуют дополнительные запросы к базе данных на этапе преобразования в доменную сущность.

Сейчас важно обращать внимание не на конкретные цифры, которые могут отличаться на разных машинах, а именно на соотношение полученных данных:

Заранее было понятно, что подходы Lazy и selectin будут последними в списке производительности, Joinload заметно превосходит их. Но безусловным лидером является CTE‑запрос в стиле core SQLAlchemy — ему нет равных, несмотря на увеличение уровня вложенности, производительность запросов остаётся стабильной. Что касается сырого CTE‑запроса, то начиная с пяти тысячи уровней вложенности наблюдается постепенное снижение скорости, и в конечном итоге он сравнивается с подходом joinload. А причина этого — функция преобразования из модели «Алхимии» в доменную сущность.

Далее было интересно покрутить параметр depth для стратегии selectin и joinload .

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

Аналогичная ситуация наблюдается и при изменении параметра depth для подхода joinload. До тысячи уровней производительность остаётся стабильной, но с дальнейшим увеличением значения скорость выполнения метода начинает ощутимо снижаться. Однако зависимость здесь не линейная: например, установка depth = 10 может положительно сказаться на производительности. Таким образом, существует возможность подобрать оптимальное значение этого параметра.

Стоит учесть, что проектирование доменной сущности, в которой атрибут является ссылкой на объект того же класса, связано с некоторыми ограничениями. Например, когда уровень вложенности превышает тысячу, интерпретатор начинает испытывать трудности в функции преобразования из модели БД SpaceModel в доменную сущность SpaceEntity, и появляется ошибка переполнения стека вызовов:

RecursionError: maximum recursion depth exceeded while getting the repr of an object

Это ограничение можно обойти, и увеличить лимит:

import sys
sys.setrecursionlimit(1500)

Для проведения нагрузочного тестирования я увеличил этот лимит, но при работе с такими большими объемами данных, полученными из базы, необходимо задуматься о том, как иначе представлять иерархию и работать с ней (подойдет и просто плоский список объектов). Однако даже после увеличения лимита возникли проблемы с pydantic. После того как репозиторий извлек данные и преобразовал их в доменную сущность, на этапе финальной сериализации в JSON pydantic выдает ошибку о входе в бесконечный цикл, хотя на самом деле это не так.

ValueError: Circular reference detected (depth exceeded)

Исправить эту ошибку не получилось: структура с глубиной вложенности свыше тысячи уровней оказалась слишком сложной для pydantic. В интернете пишут, что marshmallow умеет работать с такими данными, но я это не проверял.

Итак, при проектировании архитектуры проекта необходимо обдумывать не только способ извлечения иерархической структуры данных из базы данных, но и то, как с ней работать и в каком виде передавать другим системам. По итогам тестирования производительности можно заключить, что рекурсивный CTE-запрос является самым эффективным и универсальным решением, независимым от используемого фреймворка или языка программирования. Однако joinload тоже станет отличным выбором, если в проекте используются Python и SQLAlchemy.