如何在SQLAlchemy关联查询后构造虚拟表并映射Pydantic模型？

一、完善Pydantic响应模型C #

要让C包含A和B的所有字段且仅保留一个address，可以通过模型继承+字段重写的方式实现，避免手动重复定义字段：

from pydantic import BaseModel

# 先定义A、B各自的基础Pydantic模型
class ABase(BaseModel):
    address: str
    # 补充A的其他字段，示例：
    name: str
    age: int

    class Config:
        orm_mode = True

class BBase(BaseModel):
    address: str
    # 补充B的其他字段，示例：
    phone: str
    email: str

    class Config:
        orm_mode = True

# 合并两个模型，重写address避免字段冲突
class C(ABase, BBase):
    address: str

    class Config:
        orm_mode = True

如果A、B的字段较多，也可以通过动态生成字段自动合并（适合字段频繁变动的场景）：

from pydantic import create_model
from sqlalchemy import inspect

# 获取A、B的数据库字段类型映射
a_fields = {c.key: (c.type.python_type, ...) for c in inspect(A).mapper.column_attrs}
b_fields = {c.key: (c.type.python_type, ...) for c in inspect(B).mapper.column_attrs}

# 合并字段：保留A的address，移除B的address
combined_fields = {**a_fields, **b_fields}
combined_fields.pop('address')
combined_fields['address'] = a_fields['address']

# 动态生成Pydantic模型C
C = create_model(
    'C',
    **combined_fields,
    __config__=dict(orm_mode=True)
)

二、将SQLAlchemy查询结果转换为C实例 #

方法1：手动合并已有对象属性

如果已经通过查询得到(a,b)对象，可通过工具函数提取模型字段并合并：

from sqlalchemy import inspect

def model_to_dict(obj):
    """将SQLAlchemy模型对象转换为纯字段字典"""
    return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}

# 合并A、B的字段字典，移除B的address
a_dict = model_to_dict(a)
b_dict = model_to_dict(b)
b_dict.pop('address', None)

# 转换为C实例
c_instance = C(**{**a_dict, **b_dict})

方法2：查询时直接构造虚拟结果（更高效）

跳过对象合并步骤，直接在SQLAlchemy查询中选择所有需要的字段，返回合并后的结构：

# 选择A的所有字段 + B的非address字段
query = db.query(
    A.address, A.name, A.age,
    B.phone, B.email
).join(B, A.address == B.address)

# 获取结果并转换为C实例
result = query.first()
if result:
    c_instance = C(**result._asdict())

三、FastAPI接口中返回C实例 #

在接口中直接返回c_instance或查询结果字典，FastAPI会自动完成序列化：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

# 数据库会话依赖（示例）
def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()

@app.get("/combined-data", response_model=C)
def get_combined_data(db: Session = Depends(get_db)):
    result = db.query(A.address, A.name, A.age, B.phone, B.email).join(B, A.address == B.address).first()
    return result._asdict() if result else {}

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