FastAPI中SQLAlchemy v2多表关联查询特定格式实现问题

问题背景 #

现有三个关联模型：MaterialReceipt（父表）、MaterialReceiptLine（子表，关联父表和Material表），需要查询所有可见的物料收料单，并返回包含物料名称、SKU的嵌套格式数据。

模型定义如下：

class MaterialReceipt(Base, AuditMixins):
    material_receipt_id = Column(Integer, primary_key=True, index=True)
    pic = Column(String)
    is_visible = Column(Boolean)
    material_receipt_lines = relationship("MaterialReceiptLine", backref='material_receipt')


class MaterialReceiptLine(Base, AuditMixins):
    material_receipt_line_id = Column(Integer, primary_key=True, index=True)
    material_id = Column(Integer, ForeignKey("material.material_id"))
    material_receipt_id = Column(Integer, ForeignKey("material_receipt.material_receipt_id"))
    quantity = Column(Float)
    exp_date = Column(Date)

class Material(Base, AuditMixins):
    material_id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    sku = Column(String)
    location = Column(String)
    low_stock_threshold = Column(Float)

期望输出格式：

[
    {
        "material_receipt_id": ...,
        "pic": ...,
        "is_visible": ...,
        "material_receipt_lines": [
            {
                "material_receipt_line_id": ...,
                "name": ...,
                "sku": ...,
                "quantity": ...,
                "exp_date": ...
            }, ...
        ]
    }, ...
]

解决方案 #

方案一：模型关联+Pydantic序列化（推荐） #

1. 给MaterialReceiptLine添加与Material的关联关系

在MaterialReceiptLine模型中添加关联，让SQLAlchemy可以自动加载物料数据：

class MaterialReceiptLine(Base, AuditMixins):
    # 原有字段保持不变
    material_receipt_line_id = Column(Integer, primary_key=True, index=True)
    material_id = Column(Integer, ForeignKey("material.material_id"))
    material_receipt_id = Column(Integer, ForeignKey("material_receipt.material_receipt_id"))
    quantity = Column(Float)
    exp_date = Column(Date)
    
    # 添加关联Material的relationship
    material = relationship("Material", lazy="joined")  # 直接关联加载，避免N+1查询问题

2. 修改查询代码，加载多层关联数据

调整原有查询逻辑，通过joinedload加载MaterialReceiptLine关联的Material数据，同时用distinct()避免重复的收料单记录：

material_receipts = (
    self.db.query(MaterialReceipt)
    .join(MaterialReceiptLine)
    .options(
        contains_eager(MaterialReceipt.material_receipt_lines)
        .joinedload(MaterialReceiptLine.material)
    )
    .filter(MaterialReceipt.is_visible)
    .order_by(MaterialReceipt.created_at.desc())
    .distinct()
    .all()
)

3. 用Pydantic定义输出格式并序列化

创建Pydantic模型匹配期望的输出结构，直接从SQLAlchemy实例转换：

from pydantic import BaseModel
from datetime import date

class MaterialReceiptLineOut(BaseModel):
    material_receipt_line_id: int
    name: str
    sku: str
    quantity: float
    exp_date: date | None  # 允许日期为空

class MaterialReceiptOut(BaseModel):
    material_receipt_id: int
    pic: str | None
    is_visible: bool
    material_receipt_lines: list[MaterialReceiptLineOut]

    class Config:
        orm_mode = True  # 开启ORM模式，支持直接转换SQLAlchemy模型

# 转换查询结果为目标格式
output_data = [MaterialReceiptOut.from_orm(receipt) for receipt in material_receipts]

方案二：手动构造查询结果（无需修改模型） #

如果不想修改模型，可以直接查询所有需要的字段，手动分组构造输出：

# 查询所有所需字段
query_result = (
    self.db.query(
        MaterialReceipt.material_receipt_id,
        MaterialReceipt.pic,
        MaterialReceipt.is_visible,
        MaterialReceiptLine.material_receipt_line_id,
        Material.name,
        Material.sku,
        MaterialReceiptLine.quantity,
        MaterialReceiptLine.exp_date
    )
    .join(MaterialReceiptLine, MaterialReceipt.material_receipt_id == MaterialReceiptLine.material_receipt_id)
    .join(Material, MaterialReceiptLine.material_id == Material.material_id)
    .filter(MaterialReceipt.is_visible)
    .order_by(MaterialReceipt.created_at.desc())
    .all()
)

# 手动分组构造输出格式
output_dict = {}
for row in query_result:
    receipt_id = row.material_receipt_id
    if receipt_id not in output_dict:
        output_dict[receipt_id] = {
            "material_receipt_id": receipt_id,
            "pic": row.pic,
            "is_visible": row.is_visible,
            "material_receipt_lines": []
        }
    output_dict[receipt_id]["material_receipt_lines"].append({
        "material_receipt_line_id": row.material_receipt_line_id,
        "name": row.name,
        "sku": row.sku,
        "quantity": row.quantity,
        "exp_date": row.exp_date.isoformat() if row.exp_date else None
    })

output_data = list(output_dict.values())

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