FastAPI+SQLModel一对多嵌套表OpenAPI接口无法返回关联子数据
问题原因
三个核心配置缺失导致关联子表数据无法返回:
- 仅在Pydantic响应schema中定义了
calls字段,但标记table=True的数据库表模型未声明Relationship关系映射,ORM层不知道Customer和Call两张表的关联逻辑,无法自动挂载关联数据。 - 查询逻辑未做关联数据预加载:SQLModel底层基于SQLAlchemy,默认关联字段是懒加载模式,直接使用
db.get(Customer, id)查询时,不会主动查询关联的Call表数据;如果接口返回时数据库会话已关闭,懒加载还会直接报错,根本拿不到子表数据。 - Call表模型也未配置反向Relationship,后续要实现带客户信息的通话记录查询也会失败。
- 额外冗余问题:代码中导入了无用的
from email.policy import default,不影响功能但可以清理。
修复步骤
1. 补全表模型的Relationship配置
Relationship必须声明在标记了table=True的ORM模型上,和外键配合完成关联映射,修正后的模型代码如下:
from datetime import datetime from sqlalchemy import UniqueConstraint from sqlmodel import Field, SQLModel, Relationship class CustomerBase(SQLModel): __table_args__ = (UniqueConstraint("email"),) first_name: str last_name: str email: str active: bool | None = True class Customer(CustomerBase, table=True): id: int | None = Field(primary_key=True, default=None) # 一对多关系映射:一个客户对应多条通话记录,和Call表的customer字段双向绑定 calls: list["Call"] = Relationship(back_populates="customer") class CustomerCreate(CustomerBase): pass class CustomerRead(CustomerBase): id: int class CustomerReadWithCalls(CustomerRead): calls: list["CallRead"] = [] class CallBase(SQLModel): duration: int cost_per_minute: int | None = None customer_id: int | None = Field(default=None, foreign_key="customer.id") created: datetime = Field(nullable=False, default=datetime.now().date()) class Call(CallBase, table=True): id: int | None = Field(primary_key=True) # 多对一反向映射:一条通话记录归属一个客户 customer: Customer | None = Relationship(back_populates="calls") class CallCreate(CallBase): pass class CallRead(CallBase): id: int class CallReadWithCustomer(CallRead): customer: CustomerRead | None
2. 修改查询逻辑,预加载关联数据
懒加载模式在FastAPI接口场景下很容易因为会话关闭失效,一对多关系推荐使用selectinload做预加载,一次性查出关联数据,修正后的CRUD代码如下:
from sqlmodel import select from sqlalchemy.orm import selectinload from rbi_app.database import Session from rbi_app.models import Customer def get_customer(db: Session, id: int): # 查询客户时同步预加载关联的通话记录 stmt = select(Customer).where(Customer.id == id).options(selectinload(Customer.calls)) return db.exec(stmt).first() def get_customers(db: Session, email: str = "", offset: int = 0, limit: int = 100): if email: stmt = select(Customer).where(Customer.email == email).options(selectinload(Customer.calls)) return db.exec(stmt).first() # 列表接口当前响应模型不需要返回calls,可不用加预加载配置 stmt = select(Customer).offset(offset).limit(limit).order_by(Customer.id) return db.exec(stmt).all()
说明
- 上述修改不需要重建数据库表,Relationship是ORM层的映射逻辑,不会修改已有的表结构,只要原表的
customer_id外键约束存在即可直接生效。 - 一对多场景优先用
selectinload,会分两次SQL查询主表和子表数据,不会产生笛卡尔积,性能优于联表加载。 - 修改完成后重启服务,调用单个客户查询接口即可正常返回关联的calls列表数据,不会再出现空数组的问题。
内容的提问来源于stack exchange,提问作者writes_on
相关产品推荐
相关产品推荐

