FastAPI中Async SQLAlchemy关联模型与Pydantic兼容方案咨询

核心思路是在异步上下文内完成所有关联数据的预加载，再将完整的SQLAlchemy模型实例转换为Pydantic模型，避免在Pydantic同步验证器中执行异步操作。

一、修正SQLAlchemy模型的基础错误 #

你当前的Bar模型继承Foo是逻辑错误（两个独立表应都继承Base），且relationship类型声明有误，修正后代码如下：

foo.py（SQLAlchemy模型） #

from sqlalchemy import Integer, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Foo(Base):
    __tablename__ = "foo"
    id_: Mapped[int] = mapped_column(Integer, primary_key=True)
    bar_id: Mapped[int] = mapped_column(Integer, ForeignKey("bar.id_"))
    bar: Mapped["Bar"] = relationship("Bar", back_populates="foo")

bar.py（SQLAlchemy模型） #

from sqlalchemy import Integer, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .foo import Base

class Bar(Base):
    __tablename__ = "bar"
    id_: Mapped[int] = mapped_column(Integer, primary_key=True)
    foo_id: Mapped[int] = mapped_column(Integer, ForeignKey("foo.id_"))
    foo: Mapped["Foo"] = relationship("Foo", back_populates="bar")

二、处理Pydantic模型的循环引用 #

由于Foo和Bar互相引用，需用ForwardRef解决循环导入问题：

foo.py（Pydantic Schema） #

from pydantic import BaseModel, Field
from typing import Optional
from typing_extensions import ForwardRef

Bar = ForwardRef("Bar")

class Foo(BaseModel):
    id_: int = Field(...)
    bar_id: int = Field(...)
    bar: Optional[Bar] = None

from .bar import Bar
Foo.model_rebuild()

bar.py（Pydantic Schema） #

from pydantic import BaseModel, Field
from typing import Optional
from typing_extensions import ForwardRef

Foo = ForwardRef("Foo")

class Bar(BaseModel):
    id_: int = Field(...)
    foo_id: int = Field(...)
    foo: Optional[Foo] = None

from .foo import Foo
Bar.model_rebuild()

三、异步查询时预加载关联数据 #

在FastAPI异步路由中，用SQLAlchemy的selectinload或joinedload预加载关联，再通过from_orm直接转换为Pydantic模型：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
from .models.foo import Foo as DB_Foo
from .schemas.foo import Foo as FooSchema
from .database import get_db

app = FastAPI()

@app.get("/foo/{foo_id}", response_model=FooSchema)
async def get_foo(foo_id: int, db: AsyncSession = Depends(get_db)):
    # 预加载bar关联数据
    result = await db.execute(
        select(DB_Foo)
        .where(DB_Foo.id_ == foo_id)
        .options(selectinload(DB_Foo.bar))
    )
    db_foo = result.scalar_one_or_none()
    if not db_foo:
        raise HTTPException(status_code=404, detail="Foo not found")
    # 直接转换为Pydantic模型，此时bar已加载完成
    return FooSchema.from_orm(db_foo)

四、为什么之前的方法失效？ #

• selectinload等懒加载策略：这类加载需要在异步上下文内触发，但Pydantic验证器是同步函数，无法执行await操作，因此必须在转换模型前完成关联数据的加载。
• joinedload的字段命名问题：joinedload生成的左连接数据会被SQLAlchemy自动填充到模型的relationship属性中，使用from_orm转换时，Pydantic会直接读取bar属性，无需处理特殊命名字段。

可选优化：设置默认加载策略 #

如果希望每次查询Foo时自动加载bar，可在SQLAlchemy模型的relationship中设置默认懒加载策略：

class Foo(Base):
    # ... 其他字段
    bar: Mapped["Bar"] = relationship("Bar", back_populates="foo", lazy="selectin")

后续查询无需手动添加options(selectinload(...))，关联数据会自动异步加载。

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