异步SQLAlchemy中循环关联的Pydantic验证异常问题

我定义了SQLAlchemy模型实现Test与TestAttribute的一对多关联，TestAttribute还可关联其他Test对象（关联链数据层面保证终止），同时编写了对应Pydantic Schema和提供get（按ID查单条）、list（查全部）方法的TestRepository类。目前遇到的问题是：调用get方法时，若Test对象存在多层关联属性，Pydantic验证会抛出MissingGreenlet错误；但调用list方法时可正常返回嵌套结构数据。

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"
    )

Pydantic Schema代码 #

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)

Repository类代码 #

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_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=<src.db.test.Test object at 0x7fadc2b01b90>, input_type=Test]

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": []
  }
]

问题原因 #

核心差异在于异步会话内的数据加载时机：

• list方法查询全部Test对象时，Test.attributes的lazy="selectin"会批量加载所有关联的TestAttribute，而TestAttribute.test_attribute的lazy="joined"会直接关联加载对应的Test对象，多层嵌套的关联数据在会话打开期间就被全部加载到内存，Pydantic验证时无需触发新查询。
• get方法查询单条Test对象时，当Pydantic递归验证到test_attribute.attributes时，触发了Test.test_attributes的延迟加载（该关系未设置异步兼容的lazy策略，默认是同步的select），此时即使会话未关闭，异步环境下无greenlet上下文支持，直接抛出MissingGreenlet错误。

解决方法 #

方法1：修改模型的lazy加载策略 #

为Test.test_attributes设置异步兼容的lazy="selectin"策略，确保延迟加载在异步会话内可正常执行：

class Test(DeclarativeBase):
    # ... 其他字段不变
    test_attributes: Mapped[list["TestAttribute"]] = relationship(
        "TestAttribute",
        back_populates="test_attribute",
        primaryjoin="Test.id == TestAttribute.test_attribute_id",
        lazy="selectin"  # 添加这一行
    )

方法2：显式预加载所有嵌套关联 #

在get方法中使用SQLAlchemy的预加载API，一次性加载所有需要的嵌套关联，避免延迟加载：

from sqlalchemy.orm import selectinload, joinedload

class TestRepository:
    
    @staticmethod
    async def get(id: int) -> TestSchema | None:
        async with async_session_maker() as session:
            result = await session.execute(
                select(Test)
                .where(Test.id == id)
                .options(
                    # 预加载当前Test的attributes
                    selectinload(Test.attributes)
                    # 预加载每个attribute对应的test_attribute
                    .joinedload(TestAttribute.test_attribute)
                    # 预加载test_attribute的attributes
                    .selectinload(Test.attributes)
                    # 如需更深层关联，继续添加链式预加载
                )
            )
            data = result.scalars().one_or_none()
            return TestSchema.model_validate(data) if data else None

方法3：确保Pydantic验证在会话内完成（已满足，可配合其他方法使用） #

当前get方法已经在async with会话块内执行model_validate，但需确保所有关联加载的触发都在会话关闭前，结合方法1或2即可解决问题。

内容的提问来源于stack exchange，提问作者Кирилл Сыроежкин