FastAPI+SQLModel关联实体嵌套返回报错及数据为空问题
FastAPI+SQLModel关联接口问题修复
问题1:Swagger文档抛出TypeError: issubclass() arg 1 must be a class
故障原因
TYPE_CHECKING常量仅在静态类型检查阶段为True,代码运行时该值为False,块内的导入语句不会实际执行。将ProductRead放在TYPE_CHECKING块中导入时,运行时该类不存在,FastAPI生成OpenAPI文档、Pydantic解析模型字段时,无法识别List["ProductRead"]的实际类型,就会抛出该类型错误。
这里存在写法误区:仅带有table=True的ORM表模型(Shop、Product)会因为双向Relationship定义产生循环导入问题,ProductRead这类纯Pydantic响应模型不包含反向关系绑定,导入时不会触发循环依赖,不需要放在TYPE_CHECKING块内。
修复步骤
修改app/models/shop.py文件:
- 文件顶部添加前向引用兼容导入
- 仅将ORM表模型
Product保留在TYPE_CHECKING块内,将ProductRead移至顶层常规导入 - 模型定义完成后手动更新前向引用
修改后的shop.py模型文件核心代码参考:
from __future__ import annotations from sqlmodel import SQLModel, Field, Relationship from typing import TYPE_CHECKING, List, Optional # 仅ORM表模型放在TYPE_CHECKING块避免循环导入 if TYPE_CHECKING: from app.models.product import Product # Pydantic响应模型直接顶层导入,无循环依赖风险 from app.models.product import ProductRead class ShopBase(SQLModel): name: Optional[str] = None class Shop(ShopBase, table=True): __tablename__ = "shop" id: Optional[int] = Field(default=None, primary_key=True, nullable=False) products: List["Product"] = Relationship(back_populates="shop") class ShopRead(ShopBase): id: int class ShopGet(ShopRead): products: List[ProductRead] = [] is_main: bool = True class ShopCreate(ShopBase): pass # 模型定义完成后更新前向引用 ShopGet.update_forward_refs()
问题2:查询店铺详情时关联products列表返回空
故障原因
- SQLModel/SQLAlchemy的一对多关系默认采用懒加载策略,异步Session场景下,直接调用
session.get(Shop, shop_id)查询店铺对象时,不会自动发起关联商品表的查询 - 懒加载生成的关联属性占位符需要在Session上下文内主动触发加载,Session关闭后访问该属性只能拿到空列表,不会自动补发SQL查询关联数据
修复步骤
修改app/repositories/shop.py的get_by_id方法,使用selectinload显式预加载关联商品数据,不要直接使用session.get()查询:
- 文件顶部导入预加载工具和查询方法
- 重写查询逻辑,关联加载products数据
修改后的shop仓库代码参考:
from typing import Optional from sqlalchemy.future import select from sqlalchemy.orm import selectinload from app.models.shop import Shop, ShopCreate from app.repositories.base import BaseRepository class ShopRepository(BaseRepository): async def create(self, shop: ShopCreate) -> Shop: db_shop = Shop.from_orm(shop) self.session.add(db_shop) await self.session.commit() await self.session.refresh(db_shop) return db_shop async def get_by_id(self, shop_id: int) -> Optional[Shop]: query = select(Shop).where(Shop.id == int(shop_id)).options(selectinload(Shop.products)) result = await self.session.execute(query) return result.scalar_one_or_none()
注意该方法返回值标注为Optional[Shop]即可,不需要返回ShopGet类型:接口层已经声明response_model=ShopGet,FastAPI会自动将查询到的ORM实例序列化为对应响应结构,自动填充is_main默认值、将关联商品转为ProductRead格式,最终返回期望的响应结构。
内容的提问来源于stack exchange,提问作者Gentleman
相关产品推荐
相关产品推荐

