FastAPI Users关联模型路由因异步SQLAlchemy懒加载失败

问题描述 #

在FastAPI项目中使用FastAPI Users库时，为User模型添加与Post模型的一对多关联后，库内置的注册、获取用户等路由出现异常。具体表现为：当UserRead Pydantic模型包含posts字段时，调用POST /auth/register会触发错误：

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

移除UserRead中的posts字段后，路由恢复正常。尝试为User_DB的所有关联设置预加载，但担心这种方式会强制预加载所有关联属性，且FastAPI Users文档未提及该方案，因此寻求在FastAPI Users中正确使用关联关系的方法。

涉及代码 #

SQLAlchemy模型

class Mixin(MappedAsDataclass, SQLAlchemyBaseUserTable[int]):
    pass

class User_DB(BaseModel_DB, Mixin):
    __tablename__ = "user_table"
    id: Mapped[int] = mapped_column(primary_key=True, init=False)  # type: ignore
    firstname: Mapped[str] = mapped_column()
    posts: Mapped[list["Post_DB"]] = relationship(init=False)

class Post_DB(BaseModel_DB):
    __tablename__ = "post_table"

    id: Mapped[int] = mapped_column(primary_key=True, init=False)
    name: Mapped[str] = mapped_column()
    user_id: Mapped[int] = mapped_column(ForeignKey("user_table.id"))

Pydantic Schema

class UserRead(fastapi_users_schemas.BaseUser[int], BaseSchema):
    firstname: str
    email: str
    posts: list[PostRead]  # 注释此字段路由可正常工作

解决方案 #

1. 拆分Pydantic Schema（推荐） #

创建两套User相关的Schema：一套用于FastAPI Users内置路由（不包含关联字段），另一套用于需要返回关联数据的自定义路由。这样既能保留内置路由的正常功能，又能在需要时返回关联数据。

修改后的Schema代码

# 基础Schema，用于内置路由
class UserReadBase(fastapi_users_schemas.BaseUser[int], BaseSchema):
    firstname: str
    email: str

# 带关联字段的Schema，用于自定义路由
class UserReadWithPosts(UserReadBase):
    posts: list[PostRead]

初始化FastAPI Users时使用基础Schema

# 初始化用户管理器和认证路由
user_manager = UserManager(user_db, auth_backend)
app.include_router(
    fastapi_users.get_auth_router(auth_backend),
    prefix="/auth/jwt",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_register_router(UserReadBase, UserCreate),
    prefix="/auth",
    tags=["auth"],
)

自定义路由返回带关联的用户数据

在自定义路由中显式预加载posts关联：

@app.get("/users/{user_id}/with-posts", response_model=UserReadWithPosts)
async def get_user_with_posts(user_id: int, session: AsyncSession = Depends(get_async_session)):
    user = await session.get(User_DB, user_id, options=[selectinload(User_DB.posts)])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

2. 自定义数据库适配器，按需预加载关联 #

继承FastAPI Users的AsyncSQLAlchemyUserDatabase，重写获取用户的方法，在需要时预加载指定关联。这种方式可以让内置路由也能返回关联数据，但需要谨慎处理，避免不必要的性能开销。

自定义数据库适配器代码

from fastapi_users_db_sqlalchemy import AsyncSQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

class CustomAsyncUserDatabase(AsyncSQLAlchemyUserDatabase[int, User_DB]):
    def __init__(self, session: AsyncSession, user_model: Type[User_DB], preload_options=None):
        super().__init__(session, user_model)
        self.preload_options = preload_options or []

    async def get(self, user_id: int) -> Optional[User_DB]:
        statement = select(self.user_model).where(self.user_model.id == user_id).options(*self.preload_options)
        result = await self.session.execute(statement)
        return result.scalar_one_or_none()

# 初始化时传入预加载选项
user_db = CustomAsyncUserDatabase(session, User_DB, preload_options=[selectinload(User_DB.posts)])

3. 调整SQLAlchemy关系配置（不推荐） #

可以将关系设置为lazy="selectin"，但这种方式会在访问关联字段时自动触发异步查询，虽然能解决错误，但可能导致不可控的IO操作，增加性能隐患，不建议在生产环境使用。

class User_DB(BaseModel_DB, Mixin):
    # ... 其他字段
    posts: Mapped[list["Post_DB"]] = relationship(init=False, lazy="selectin")

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