FastAPI+SQLAlchemy查询结果结构不符，需调整为平级格式求助

我使用Python的FastAPI和SQLAlchemy开发API，已完成请求与响应的Pydantic模型校验，但当前响应结构不符合预期，问题出在SQLAlchemy的查询返回方式上。

最初匹配自定义Pydantic模型时遇到阻碍，后发现需让模型属性名与查询的表名一致才能匹配。当前模型定义如下：

from datetime import datetime
from pydantic import BaseModel

class BasePost(BaseModel):
    title: str
    content: str
    published: bool = True
    
class ExtraPost(BasePost):
    creation_date: datetime
    last_update: datetime
    user_id: int
    user: ResponseUser

class ResponsePost(BaseModel):
    PostsTable: ExtraPost
    vote_count: int

当前使用ResponsePost作为FastAPI接口的response_model虽能正常返回结果，但并非我想要的结构。期望的模型结构为：

from datetime import datetime
from pydantic import BaseModel

class BasePost(BaseModel):
    title: str
    content: str
    published: bool = True 

class ResponsePost(BaseModel):
    title: str
    content: str
    published: bool = True
    creation_date: datetime
    last_update: datetime
    user_id: int
    user: ResponseUser
    vote_count: int

对应的接口代码：

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import func
from your_module import PostsTable, VotesTable, get_user_from_token, get_db, INTEGER_LIMIT, ResponseUser

router = APIRouter()

@router.get("/", response_model=List[ResponsePost])
def read_posts(user=Depends(get_user_from_token), db: Session = Depends(get_db), limit: int = INTEGER_LIMIT, offset: int = 0):
    try:
        post_query = db.query(PostsTable, func.count(VotesTable.post_id).label("vote_count")).\
            outerjoin(VotesTable, PostsTable.id == VotesTable.post_id).\
            group_by(PostsTable.id).\
            limit(limit).offset(offset).all()
    except Exception as e:
        print(e)
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
    return post_query

调试发现，SQLAlchemy返回的对象包含表模型名称，导致结果为PostsTable嵌套结构，我需要PostsTable的所有属性与vote_count处于同一层级，仅user属性保留嵌套对象。

解决方案 #

方法1：转换查询结果为扁平化字典 #

在返回结果前，将SQLAlchemy查询返回的元组（PostsTable实例 + vote_count）转换为扁平化字典，让Pydantic直接映射到期望的ResponsePost模型：

@router.get("/", response_model=List[ResponsePost])
def read_posts(user=Depends(get_user_from_token), db: Session = Depends(get_db), limit: int = INTEGER_LIMIT, offset: int = 0):
    try:
        post_query = db.query(PostsTable, func.count(VotesTable.post_id).label("vote_count")).\
            outerjoin(VotesTable, PostsTable.id == VotesTable.post_id).\
            group_by(PostsTable.id).\
            limit(limit).offset(offset).all()
        
        # 转换为扁平化结构
        flattened_results = []
        for post_obj, vote_count in post_query:
            # 将SQLAlchemy实例转为字典，移除内部属性
            post_dict = {k: v for k, v in post_obj.__dict__.items() if not k.startswith("_")}
            # 合并vote_count到同一层级
            post_dict["vote_count"] = vote_count
            flattened_results.append(post_dict)
            
    except Exception as e:
        print(e)
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
    return flattened_results

方法2：通过Pydantic别名映射字段 #

若不想修改查询结果处理逻辑，可给ResponsePost的字段添加别名，让其匹配SQLAlchemy返回的嵌套结构：

from pydantic import BaseModel, Field
from datetime import datetime

class ResponsePost(BaseModel):
    title: str = Field(alias="PostsTable.title")
    content: str = Field(alias="PostsTable.content")
    published: bool = Field(alias="PostsTable.published")
    creation_date: datetime = Field(alias="PostsTable.creation_date")
    last_update: datetime = Field(alias="PostsTable.last_update")
    user_id: int = Field(alias="PostsTable.user_id")
    user: ResponseUser = Field(alias="PostsTable.user")
    vote_count: int

    class Config:
        populate_by_name = True  # 允许通过别名填充字段

此方法需逐个字段配置别名，适合字段较少的场景。

方法3：直接查询所需字段（SQLAlchemy with_entities） #

若无需返回PostsTable的全部字段，可直接指定查询字段，让SQLAlchemy返回扁平化元组，Pydantic会自动按顺序映射：

@router.get("/", response_model=List[ResponsePost])
def read_posts(user=Depends(get_user_from_token), db: Session = Depends(get_db), limit: int = INTEGER_LIMIT, offset: int = 0):
    try:
        post_query = db.query(
            PostsTable.title,
            PostsTable.content,
            PostsTable.published,
            PostsTable.creation_date,
            PostsTable.last_update,
            PostsTable.user_id,
            PostsTable.user,
            func.count(VotesTable.post_id).label("vote_count")
        ).outerjoin(VotesTable, PostsTable.id == VotesTable.post_id).\
            group_by(PostsTable.id).\
            limit(limit).offset(offset).all()
    except Exception as e:
        print(e)
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
    return post_query

注意：查询字段的顺序需与ResponsePost的字段顺序完全一致。

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