Проблема с циклическим взаимодействием в асинхронной SQLAlchemy ⇐ Python

[Anonymous]

У меня есть следующая структура sqlalchemy для некоторых объектов:
from typing import Optional

from sqlalchemy import ForeignKey, Integer
from sqlalchemy.orm import DeclarativeBase, mapped_column, relationship
from sqlalchemy.orm.base import Mapped

class Test(DeclarativeBase):
__tablename__ = "test"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
value: Mapped[int] = mapped_column(Integer, nullable=False)
attributes: Mapped[list["TestAttribute"]] = relationship(
"TestAttribute", back_populates="test", primaryjoin="Test.id == TestAttribute.test_id", lazy="selectin"
)
test_attributes: Mapped[list["TestAttribute"]] = relationship(
"TestAttribute",
back_populates="test_attribute",
primaryjoin="Test.id == TestAttribute.test_attribute_id",
)

class TestAttribute(DeclarativeBase):
__tablename__ = "test_attribute"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
test_value: Mapped[int] = mapped_column(Integer, nullable=False)
test_id: Mapped[int] = mapped_column(ForeignKey("test.id"), nullable=False)
test_attribute_id: Mapped[Optional[int]] = mapped_column(ForeignKey("test.id"), nullable=True)
test: Mapped["Test"] = relationship(back_populates="attributes", foreign_keys=[test_id])
test_attribute: Mapped[Optional["Test"]] = relationship(
back_populates="test_attributes", foreign_keys=[test_attribute_id], lazy="joined"
)


Существует объект Test с атрибутами TestAttribute. Они связаны один со многими. Кроме того, TestAttribute связан с другим объектом Test (который также может иметь свои собственные атрибуты). У меня также есть следующие понятные схемы для работы с этими моделями:
from pydantic import BaseModel, ConfigDict
from typing import Optional

class TestAttributeSchema(BaseModel):
test_attribute_id: Optional[int]
test_value: int
test_attribute: Optional["TestSchema"]
model_config = ConfigDict(from_attributes=True)

class TestSchema(BaseModel):
id: int
value: int
attributes: list[TestAttributeSchema]
model_config = ConfigDict(from_attributes=True)

Существует также следующий класс репозитория для получения тестового объекта по идентификатору:
class TestRepository:

@staticmethod
async def get(id: int) -> TestSchema:
async with async_session_maker() as session:
result = await session.execute(
select(Test).where(Test.id == id)
)
data = result.scalars().one_or_none()
if data:
return TestSchema.model_validate(data)
return None

@staticmethod
async def list() -> list[TestSchema]:
async with async_session_maker() as session:
result = await session.execute(select(Test))
return [TestSchema.model_validate(i) for i in result.scalars().all()]

Проблема: если у тестового объекта есть атрибуты, которые также ссылаются на тестовые объекты с атрибутами, которые также имеют ссылки на тестовые объекты и т.д. (на уровне данных гарантируется сходимость этой последовательности !), то при проверке с помощью pydantic при вызове метода TestRepository.get я получаю следующую ошибку:
pydantic_core._pydantic_core.ValidationError: 1 validation error for TestSchema
attributes.0.test_attribute.attributes
Error extracting attribute: 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) [type=get_attribute_error, input_value=, input_type=Test]

Самое забавное, что при вызове метода TestRepository.list такой проблемы нет! Все работает так, как задумано.
Пример правильно выполняемой программы в случае со списком:
[
{
"id": 1,
"value": 1,
"attributes": [
{
"test_attribute_id": 2,
"test_value": 1,
"test_attribute": {
"id": 2,
"value": 2,
"attributes": [
{
"test_attribute_id": 3,
"test_value": 1,
"test_attribute": {
"id": 3,
"value": 3,
"attributes": [
{
"test_attribute_id": 4,
"test_value": 1,
"test_attribute": {
"id": 4,
"value": 4,
"attributes": []
}
}
]
}
}
]
}
},
{
"test_attribute_id": 3,
"test_value": 1,
"test_attribute": {
"id": 3,
"value": 3,
"attributes": [
{
"test_attribute_id": 4,
"test_value": 1,
"test_attribute": {
"id": 4,
"value": 4,
"attributes": []
}
}
]
}
}
]
},
{
"id": 2,
"value": 2,
"attributes": [
{
"test_attribute_id": 3,
"test_value": 1,
"test_attribute": {
"id": 3,
"value": 3,
"attributes": [
{
"test_attribute_id": 4,
"test_value": 1,
"test_attribute": {
"id": 4,
"value": 4,
"attributes": []
}
}
]
}
}
]
},
{
"id": 3,
"value": 3,
"attributes": [
{
"test_attribute_id": 4,
"test_value": 1,
"test_attribute": {
"id": 4,
"value": 4,
"attributes": []
}
}
]
},
{
"id": 4,
"value": 4,
"attributes": []
},
{
"id": 5,
"value": 5,
"attributes": []
},
{
"id": 6,
"value": 6,
"attributes": []
}
]


Подробнее здесь: https://stackoverflow.com/questions/787 ... sqlalchemy