使用SQLAlchemy AsyncSession时FastAPI模型验证报错求助

问题背景 #

我基于FastAPI搭建服务，用SQLAlchemy异步连接PostgreSQL，遵循官方异步最佳实践配置了数据库依赖。创建用户的CRUD操作完成后，尝试用Pydantic的UserResponseSchema验证返回的User对象时，触发MissingGreenlet错误，涉及custom_apps和payment_info这两个懒加载关联字段。改用FastAPI的response_model参数也会出现相同原因的500错误，希望在保留懒加载机制的前提下解决该问题。

相关代码 #

数据库依赖配置 #

import os
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, AsyncAttrs
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import DeclarativeBase
from typing import AsyncGenerator
from typing import Generator
from urllib.parse import quote

# 设置DATABASE_URL常量...

engine = create_async_engine(DATABASE_URL, echo=True)

# 配置异步SessionLocal类
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession, expire_on_commit=False)

class Base(AsyncAttrs, DeclarativeBase):
    pass

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as db:
        yield db

CRUD创建用户代码 #

async def create_user(db: AsyncSession, user: UserSchema) -> User:
    hashed_password = pwd_context.hash(user.password)
    db_user = User(username=user.username, email=user.email, phone_number=user.phone_number, hashed_password=hashed_password)
    db.add(db_user)
    try:
        await db.commit()
        await db.refresh(db_user)
    except IntegrityError as e:
        await db.rollback()
        if "unique constraint" in str(e.orig).lower():
            raise UserExistsException("User with given email or username already exists.")
        else:
            raise
    return db_user

路由代码 #

from ..crud.users import create_user as crud_create_user
# 其他导入...

@router.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserSchema, db: AsyncSession = Depends(get_db)):
    """
    创建新用户
    """
    try:
        user = await crud_create_user(db=db, user=user)
        print(f"{user = }")
        print(f"{user.__dict__ = }")
        print("This works")
        
        # 问题所在：无法提取属性
        user = UserResponseSchema.model_validate(user)

        return user
    except (ValueError, UserExistsException) as e:
        raise HTTPException(status_code=400, detail=str(e))

User模型定义 #

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncAttrs
import enum
from sqlalchemy import Integer, String, Boolean, Enum
from sqlalchemy.orm import Mapped, mapped_column, relationship

class Base(AsyncAttrs, DeclarativeBase):
    pass

class UserRole(enum.Enum):
    USER = "user"
    ADMIN = "admin"

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String, unique=True)
    email: Mapped[str] = mapped_column(String, unique=True)
    phone_number: Mapped[str] = mapped_column(String, unique=True)
    hashed_password: Mapped[str] = mapped_column(String)
    role: Mapped[UserRole] = mapped_column(Enum(UserRole, name='role'), default=UserRole.USER)
    is_email_verified: Mapped[bool] = mapped_column(Boolean, default=False)
    is_phone_verified: Mapped[bool] = mapped_column(Boolean, default=False)
    
    custom_apps: Mapped["CustomApp"] = relationship("CustomApp", back_populates='owner')
    payment_info: Mapped["PaymentInfo"] = relationship("PaymentInfo", back_populates='user')
    support_tickets: Mapped["SupportTicket"] = relationship("SupportTicket", back_populates='user')

错误信息 #

2 validation errors for UserResponseSchema
custom_apps
  Error extracting attribute: MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? [type=get_attribute_error, input_value=<chatbot_deployment_platf...t at 0x0000027ABB8F9E10>, input_type=User]
payment_info
  Error extracting attribute: MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? [type=get_attribute_error, input_value=<chatbot_deployment_platf...t at 0x0000027ABB8F9E10>, input_type=User]

错误原因 #

核心问题是Pydantic验证时尝试访问未加载的懒加载关联字段，而此时数据库会话已经关闭。虽然设置了expire_on_commit=False，但当代码退出get_db的async with上下文后，会话已关闭，无法触发SQLAlchemy的异步懒加载查询，因此抛出MissingGreenlet错误。

解决方案（保留懒加载机制） #

以下几种方法都可以解决问题，同时保留懒加载的默认行为：

1. 修改Pydantic响应模型，排除不需要的懒加载字段 #

如果响应中不需要返回custom_apps、payment_info这类关联字段，直接在响应模型中排除它们，或者在验证时手动指定排除：

# 方法1：定义响应模型时只包含需要的字段
class UserResponseSchema(BaseModel):
    id: int
    username: str
    email: str
    phone_number: str
    role: UserRole
    is_email_verified: bool
    is_phone_verified: bool

# 方法2：验证时手动排除关联字段
user = UserResponseSchema.model_validate(user, exclude={"custom_apps", "payment_info", "support_tickets"})

2. 在会话内预加载需要返回的关联字段 #

如果确实需要返回关联字段，在CRUD操作的会话生命周期内（会话未关闭前）预加载这些字段，避免后续触发懒加载：

# 修改CRUD的create_user函数，refresh时预加载需要的字段
async def create_user(db: AsyncSession, user: UserSchema) -> User:
    hashed_password = pwd_context.hash(user.password)
    db_user = User(username=user.username, email=user.email, phone_number=user.phone_number, hashed_password=hashed_password)
    db.add(db_user)
    try:
        await db.commit()
        # 预加载需要返回的关联字段，其他字段仍保持懒加载
        await db.refresh(db_user, attribute_names=["custom_apps", "payment_info"])
    except IntegrityError as e:
        await db.rollback()
        if "unique constraint" in str(e.orig).lower():
            raise UserExistsException("User with given email or username already exists.")
        else:
            raise
    return db_user

3. 利用AsyncAttrs异步加载需要的字段 #

由于你的Base类继承了AsyncAttrs，可以在会话未关闭时，手动异步加载需要的关联字段：

# 在路由中，会话还未关闭时加载字段
@router.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserSchema, db: AsyncSession = Depends(get_db)):
    try:
        user = await crud_create_user(db=db, user=user)
        # 异步加载需要的关联字段，其他字段仍保持懒加载
        await user.awaitable_attrs.custom_apps
        await user.awaitable_attrs.payment_info
        user = UserResponseSchema.model_validate(user)
        return user
    except (ValueError, UserExistsException) as e:
        raise HTTPException(status_code=400, detail=str(e))

4. 用Pydantic字段验证器处理未加载的关联 #

通过Pydantic的字段验证器，识别未加载的关联字段并返回默认值（如None），避免触发懒加载查询：

from pydantic import BaseModel, field_validator
from sqlalchemy.orm import Unloaded

class UserResponseSchema(BaseModel):
    id: int
    username: str
    email: str
    custom_apps: list["CustomAppSchema"] | None = None
    payment_info: "PaymentInfoSchema" | None = None

    @field_validator("custom_apps", "payment_info", mode="before")
    def handle_unloaded_fields(cls, value):
        if isinstance(value, Unloaded):
            return None
        return value

# 按需导入关联模型的Schema
from .custom_app import CustomAppSchema
from .payment_info import PaymentInfoSchema
UserResponseSchema.model_rebuild()

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