异步SQLAlchemy+Strawberry GraphQL递归多对多关系报错排查

问题场景 #

定义了User和Class的多对多关系，通过GraphQL查询用户信息时触发以下错误：

sqlalchemy.exc.StatementError: (sqlalchemy.exc.MissingGreenlet) greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place?

已尝试修改SQLAlchemy关系的lazy属性为selectin、joined、subquery，但问题未解决。

核心原因 #

在同步的__init__方法中直接访问ORM关联属性（如user_info.classes、class_info.students），而异步SQLAlchemy的延迟加载需要在greenlet_spawn管理的异步上下文中执行IO操作，同步方法没有该上下文，因此触发报错。

解决方案 #

方案一：用Strawberry字段解析器异步加载关联数据 #

将关联字段的加载逻辑从__init__移到异步解析器中，确保IO操作在异步上下文执行：

@strawberry.type(description="Class type")
class Class:
    id: int = strawberry.field(description="Class ID(Auto Increment)")
    name: str
    students: list[Annotated["User", strawberry.lazy("graphql_route.user")]] = strawberry.field(
        description="Student list",
        resolver=lambda root, info: Class.resolve_students(root, info)
    )

    def __init__(self, class_info: ClassInfo):
        self.id = class_info.id
        self.name = class_info.name

    @staticmethod
    async def resolve_students(root: "Class", info: Info) -> list["User"]:
        session: AsyncSession = info.context.db
        class_info = await session.get(
            db_models.ClassInfo, 
            root.id, 
            options=[selectinload(db_models.ClassInfo.students)]
        )
        return [User(student) for student in class_info.students]

@strawberry.type
class User:
    id: str = strawberry.field(description="User UUID(128-bit hex string)")
    email: str
    name: str
    userType: UserType
    grade: Grade | None
    school: str | None
    grade: int | None
    phone: str | None
    parentPhone: str | None
    profileImage: str | None = strawberry.field(description="Profile image URL")
    hasProfileImage: bool = strawberry.field(description="Whether the user has a profile image")
    classes: typing.List['Class'] = strawberry.field(
        description="Class list",
        resolver=lambda root, info: User.resolve_classes(root, info)
    )

    def __init__(self, user_info: db_models.UserInfo):
        print(user_info.id)
        self.id = binascii.b2a_hex(user_info.id).decode()
        self.email = user_info.email
        self.name = user_info.name
        self.userType = UserType(user_info.user_type)
        self.school = user_info.school
        self.grade = Grade(user_info.grade) if user_info.grade else None
        self.phone = user_info.phone
        self.parentPhone = user_info.parent_phone
        self.profileImage = user_info.profile_image
        self.hasProfileImage = bool(user_info.profile_image)

    @staticmethod
    async def resolve_classes(root: "User", info: Info) -> list["Class"]:
        session: AsyncSession = info.context.db
        user_info = await session.get(
            db_models.UserInfo, 
            binascii.a2b_hex(root.id), 
            options=[selectinload(db_models.UserInfo.classes)]
        )
        return [Class(class_info) for class_info in user_info.classes]

方案二：查询用户时预先加载关联关系 #

在查询UserInfo时，通过selectinload预先加载classes关系，确保关联数据在异步上下文内加载完成，避免后续同步访问触发延迟加载：

@strawberry.type
class Query:
    @strawberry.field(name="user", description="Get a single user", permission_classes=[AssistantPermission])
    async def single_user(self, info: Info, user_id: str) -> User | None:
        session: AsyncSession = info.context.db
        user_info = await session.execute(
            select(db_models.UserInfo)
            .where(db_models.UserInfo.id == binascii.a2b_hex(user_id))
            .options(selectinload(db_models.UserInfo.classes))  # 预先加载关联的classes
        )
        user_info = user_info.scalar()
        return User(user_info=user_info) if user_info else None

注意：如果GraphQL查询需要Class的students字段，需同样对Class的查询添加预加载，或用解析器异步加载，否则会触发相同错误。

关键提示 #

• 异步SQLAlchemy中，所有涉及ORM关联的IO操作必须在async def定义的异步函数中执行，不能在同步方法触发延迟加载。
• 预加载适合一次性加载主数据及关联数据，字段解析器更贴合GraphQL按需获取数据的特性。

内容的提问来源于stack exchange，提问作者hegelty