I have the following sqlalchemy structure for some objects:
class Test(Base):
__tablename__ = "test"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
attributes: Mapped[list["TestAttribute"]] = relationship(
"TestAttribute", back_populates="test", primaryjoin="Test.id == TestAttribute.test_id", lazy="selectin"
)
test_attributes: Mapped[list["DimensionAttribute"]] = relationship(
"TestAttribute",
back_populates="test_attribute",
primaryjoin="Test.id == TestAttribute.test_attribute_id",
)
class TestAttribute(Base):
__tablename__ = "test_attribute"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
test_value: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
test_id: Mapped[int] = mapped_column(ForeignKey("test.id"), nullable=False)
test_attribute_id: Mapped[Optional[int]] = mapped_column(ForeignKey("test.id"), nullable=False)
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"
)
There is a Test object that has TestAttribute attributes. They are connected one to many. Also, TestAttribute is associated with another Test object (which may also have its own attributes). I also have the following pydantic schemes for working with these models:
class TestAttributeSchema(BaseModel):
id: int
test_value: int
test_attribute: Optional["TestSchema"]
class TestSchema(BaseModel):
id: int
attributes: list["TestAttributeSchema"]
There is also the following repository class for getting a Test object by id:
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
Problem: if the test object has attributes that also refer to test objects with attributes that also have links to test objects, etc. (at the data level, it is guaranteed that this sequence converges!), then I get the following error when validating with pydantic: “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)”