FastAPI中Pydantic与SQLAlchemy映射问题：角色统一与空值隐藏

首先假设你的SQLAlchemy模型结构如下（如果和实际不符，只需调整字段名即可）：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class Customer(Base):
    __tablename__ = "customers"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # 其他业务字段

class History(Base):
    __tablename__ = "histories"
    id = Column(Integer, primary_key=True)
    customer_id_1 = Column(Integer, ForeignKey("customers.id"), nullable=True)
    customer_role_1 = Column(String, nullable=True)
    customer_id_2 = Column(Integer, ForeignKey("customers.id"), nullable=True)
    customer_role_2 = Column(String, nullable=True)
    customer_id_3 = Column(Integer, ForeignKey("customers.id"), nullable=True)
    customer_role_3 = Column(String, nullable=True)

    customer_1 = relationship("Customer", foreign_keys=[customer_id_1])
    customer_2 = relationship("Customer", foreign_keys=[customer_id_2])
    customer_3 = relationship("Customer", foreign_keys=[customer_id_3])

方式一：保留原customer_1/2/3字段，注入角色并隐藏Null值 #

这种方式严格保留你原有的字段结构，仅注入角色并移除Null字段：

from pydantic import BaseModel, model_serializer
from typing import Optional, Dict

class CustomerSchema(BaseModel):
    id: int
    name: str
    customer_role: str  # 统一的角色字段

    class Config:
        from_attributes = True

class HistoryResponseSchema(BaseModel):
    id: int
    customer_1: Optional[CustomerSchema] = None
    customer_2: Optional[CustomerSchema] = None
    customer_3: Optional[CustomerSchema] = None
    # 声明需要访问的role字段，否则Pydantic不会加载
    customer_role_1: Optional[str] = None
    customer_role_2: Optional[str] = None
    customer_role_3: Optional[str] = None

    @model_serializer
    def serialize(self) -> Dict[str, any]:
        # 先移除所有值为None的字段
        serialized = self.model_dump(exclude_none=True)
        # 为每个存在的customer字段注入对应的role
        for idx in [1,2,3]:
            customer_key = f"customer_{idx}"
            if customer_key in serialized:
                serialized[customer_key]["customer_role"] = serialized.pop(f"customer_role_{idx}")
        return serialized

    class Config:
        from_attributes = True

在FastAPI路由中使用时，记得预加载关联的Customer数据避免N+1查询：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session, joinedload
from your_module import History, get_db  # 替换为你的实际模块

app = FastAPI()

@app.get("/history/{history_id}", response_model=HistoryResponseSchema)
def fetch_history(history_id: int, db: Session = Depends(get_db)):
    history = db.query(History).options(
        joinedload(History.customer_1),
        joinedload(History.customer_2),
        joinedload(History.customer_3)
    ).filter(History.id == history_id).first()
    return history

方式二：转为统一的customers列表（更简洁的结构） #

如果允许调整响应结构，把所有非空的Customer合并为一个列表会更简洁：

from pydantic import BaseModel, field_validator
from typing import List

class CustomerSchema(BaseModel):
    id: int
    name: str
    customer_role: str

    class Config:
        from_attributes = True

class HistoryResponseSchema(BaseModel):
    id: int
    customers: List[CustomerSchema] = []

    @field_validator("customers", mode="before")
    def assemble_customers(cls, _, values):
        customers = []
        # 遍历三个customer字段，组装带角色的CustomerSchema
        for idx in [1,2,3]:
            customer = values.get(f"customer_{idx}")
            role = values.get(f"customer_role_{idx}")
            if customer:
                # 把SQLAlchemy实例转为字典，注入角色后实例化Schema
                customer_data = customer.__dict__.copy()
                customer_data["customer_role"] = role
                customers.append(CustomerSchema(**customer_data))
        return customers

    class Config:
        from_attributes = True
        # 包含需要访问的字段
        include = {"id", "customer_1", "customer_2", "customer_3", "customer_role_1", "customer_role_2", "customer_role_3"}

关键说明 #

1. exclude_none=True：在model_dump时自动移除值为None的字段，实现隐藏Null的需求。
2. model_serializer/field_validator：自定义字段映射逻辑，将customer_role_1/2/3的值注入到对应的CustomerSchema的customer_role字段中。
3. 预加载关联数据：使用joinedload确保查询History时一次性加载所有关联的Customer数据，避免性能问题。

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