如何在FastAPI中实现返回两个模型SQL Union结果的接口？

核心思路 #

因为Deposits和Withdrawals模型字段完全一致，只需两步实现：

1. 定义统一的响应模型，确保返回结构一致
2. 通过数据库查询（ORM或原生SQL）执行UNION/UNION ALL操作，合并结果后返回

具体实现步骤 #

1. 定义数据库模型（SQLAlchemy示例） #

假设你已经通过SQLAlchemy定义了两个表模型：

from sqlalchemy import Column, Integer, DateTime, Float
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Deposits(Base):
    __tablename__ = "deposits"
    id = Column(Integer, primary_key=True, index=True)
    created_at = Column(DateTime)
    updated_at = Column(DateTime)
    value = Column(Float)

class Withdrawals(Base):
    __tablename__ = "withdrawals"
    id = Column(Integer, primary_key=True, index=True)
    created_at = Column(DateTime)
    updated_at = Column(DateTime)
    value = Column(Float)

2. 定义统一响应模型（Pydantic） #

创建一个共用的Pydantic模型，用于标准化接口返回格式：

from pydantic import BaseModel
from datetime import datetime

class TransactionResponse(BaseModel):
    id: int
    created_at: datetime
    updated_at: datetime
    value: float

    class Config:
        orm_mode = True  # 支持直接从SQLAlchemy ORM对象转换

3. 实现合并查询接口 #

方式一：用SQLAlchemy ORM执行Union

通过ORM构建union或union_all查询，无需手写原生SQL：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import union_all, select
from your_module import Deposits, Withdrawals, TransactionResponse, get_db  # 替换成你的模块路径

app = FastAPI()

@app.get("/transactions", response_model=list[TransactionResponse])
def get_unified_transactions(db: Session = Depends(get_db)):
    # 构建两个表的字段查询
    deposits_query = select(Deposits.id, Deposits.created_at, Deposits.updated_at, Deposits.value)
    withdrawals_query = select(Withdrawals.id, Withdrawals.created_at, Withdrawals.updated_at, Withdrawals.value)
    
    # 合并查询：union_all对应UNION ALL（保留重复），union对应UNION（自动去重）
    unified_query = union_all(deposits_query, withdrawals_query)
    
    # 可选：添加排序，比如按创建时间倒序
    unified_query = unified_query.order_by(Deposits.created_at.desc())
    
    # 执行查询并转换为响应模型
    results = db.execute(unified_query).all()
    return [TransactionResponse(id=row[0], created_at=row[1], updated_at=row[2], value=row[3]) for row in results]

方式二：用原生SQL执行Union

如果更习惯手写SQL，可以直接执行原生语句：

@app.get("/transactions", response_model=list[TransactionResponse])
def get_unified_transactions(db: Session = Depends(get_db)):
    sql = """
        SELECT id, created_at, updated_at, value FROM deposits
        UNION ALL
        SELECT id, created_at, updated_at, value FROM withdrawals
        ORDER BY created_at DESC
    """
    results = db.execute(sql).mappings().all()  # mappings()返回字典格式，方便直接转模型
    return [TransactionResponse(**row) for row in results]

关键注意事项 #

• 字段一致性：必须保证两个查询的字段数量、顺序、数据类型完全匹配，否则数据库会抛出错误
• 去重选择：UNION会自动去重所有字段完全相同的记录，UNION ALL保留所有结果（性能更高，适合不需要去重的场景）
• 排序处理：排序语句要放在整个合并查询的最后，确保对合并后的结果生效

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