Рекурсивные запросы в SQLAlchemy

Оглавление

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.