FastAPI中用泛型ApiWrapper封装返回数据遇OpenAPI及关联字段问题

问题1：OpenAPI文档中data字段显示为string而非User结构 #

原因：接口返回的是自定义普通类DbWrapper的实例，而非ApiWrapper[User]实例。FastAPI在处理response_model时，无法将非Pydantic对象正确映射到泛型模型，导致Schema推断错误。

问题2：关联的Items字段无法展示 #

原因有两点：

1. SQLAlchemy默认对关联字段采用懒加载，当Pydantic序列化ORM对象时，数据库会话已关闭（get_session的yield逻辑导致返回后会话关闭），无法加载关联数据。
2. 返回的DbWrapper未启用ORM模式，无法将SQLAlchemy模型自动转换为Pydantic模型结构。

具体修复步骤 #

1. 移除无用的DbWrapper类 #

直接使用ApiWrapper泛型模型封装返回数据，无需自定义普通类。

2. 修改泛型模型的使用方式 #

确保接口返回ApiWrapper[T]的实例，而非其他对象。同时，为SQLAlchemy查询添加关联字段预加载，避免懒加载导致的会话关闭问题。

3. 修正代码后的完整示例 #

from pydantic import BaseModel
from typing import TypeVar, Generic
from fastapi import Depends, FastAPI
from sqlalchemy.ext.declarative import declarative_base
import sqlalchemy as sa
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session, relationship, joinedload

Base = declarative_base()

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()


class DbUser(Base):
    __tablename__ = 'user'

    id = sa.Column(sa.Integer(), autoincrement=True, primary_key=True)
    fname = sa.Column(sa.String(100), nullable=False)
    items = relationship("DbItem")


class DbItem(Base):
    __tablename__ = 'item'

    id = sa.Column(sa.Integer(), autoincrement=True, primary_key=True)
    user_id = sa.Column(sa.Integer(), sa.ForeignKey('user.id'), nullable=False)
    name = sa.Column(sa.String(100), nullable=False)


app = FastAPI()
T = TypeVar('T')


class ApiWrapper(BaseModel, Generic[T]):
    code: int = 1
    msg: str = "success"
    data: T

    class Config:
        orm_mode = True


class Item(BaseModel):
    id: int
    user_id: int
    name: str

    class Config:
        orm_mode = True

    class Meta:
        orm_model = DbItem


class User(BaseModel):
    id: int
    fname: str
    items: list[Item]

    class Config:
        orm_mode = True


Base.metadata.create_all(bind=engine)


@app.get("/users/{id}", response_model=ApiWrapper[User])
def get_user(id: int, session: Session = Depends(get_session)):
    # 使用joinedload预加载关联的items字段，避免懒加载时会话已关闭
    db_user = session.query(DbUser).options(joinedload(DbUser.items)).get(id)
    # 直接返回ApiWrapper实例，传入ORM对象会自动通过orm_mode转换为Pydantic模型
    return ApiWrapper(data=db_user)


@app.post("/users/")
def add_user(user: User, session: Session = Depends(get_session)):
    parsed_schema = parse_pydantic_schema(user)
    session.add(DbUser(**parsed_schema))
    session.commit()
    return ApiWrapper(code=1, msg="用户创建成功")


@app.get("/users/", response_model=ApiWrapper[list[User]])
def list_user(session: Session = Depends(get_session)):
    db_users = session.query(DbUser).options(joinedload(DbUser.items)).all()
    return ApiWrapper(data=db_users)


def is_pydantic(obj: object):
    """Checks whether an object is pydantic."""
    return type(obj).__class__.__name__ == "ModelMetaclass"


def parse_pydantic_schema(schema):
    """
        Iterates through pydantic schema and parses nested schemas
        to a dictionary containing SQLAlchemy models.
        Only works if nested schemas have specified the Meta.orm_model.
    """
    parsed_schema = dict(schema)
    for key, value in parsed_schema.items():
        try:
            if isinstance(value, list) and len(value):
                if is_pydantic(value[0]):
                    parsed_schema[key] = [item.Meta.orm_model(**item.dict()) for item in value]
            else:
                if is_pydantic(value):
                    parsed_schema[key] = value.Meta.orm_model(**value.dict())
        except AttributeError:
            raise AttributeError("Found nested Pydantic model but Meta.orm_model was not specified.")
    return parsed_schema

关键修复点说明 #

• 泛型模型正确映射：接口直接返回ApiWrapper[data=db_user]实例，FastAPI能正确识别泛型参数User，OpenAPI文档中data字段会展示User的完整Schema。
• 关联字段预加载：使用joinedload(DbUser.items)在查询用户时一次性加载关联的Item数据，避免会话关闭后无法懒加载的问题，确保Pydantic能获取到items字段的值。
• ORM模式生效：ApiWrapper和User/Item模型均启用orm_mode=True，支持直接将SQLAlchemy ORM对象转换为Pydantic模型结构。

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