SQLAlchemy查询返回嵌套结构而非预期扁平字段的问题求助

问题原因分析 #

你遇到的嵌套结构返回问题，核心原因在于**result.mappings().all()的处理逻辑**和查询整个模型对象的方式：

当使用select(cls.model)查询完整的SQLAlchemy模型实例时，result.mappings()会把每个查询结果封装成以**模型类名（这里是"Products"）**为键的映射对象，后续如果直接将这些映射返回给API（比如FastAPI自动序列化），就会保留这种嵌套结构。

解决方案（按优先级推荐） #

方案1：使用Pydantic模型序列化（最推荐）

这是规范度、可维护性最高的方案，通过Pydantic定义预期返回结构，自动完成模型实例到扁平字典的转换，同时支持类型校验和自动类型转换。

步骤如下：

1. 定义Pydantic响应模型，与数据库模型字段一一对应：

from pydantic import BaseModel, ConfigDict

class ProductResponse(BaseModel):
    # 开启从ORM实例自动转换的配置
    model_config = ConfigDict(from_attributes=True)
    
    product_id: int
    name: str
    price: int
    description: str | None  # 对应数据库允许为空的字段
    stars: int | None
    ingredients: str
    nutrition: str
    image_id: int | None

2. 修改BaseService的find_all方法，用Pydantic序列化结果：

class BaseService:
    model = None
    response_model = None  # 新增：关联响应模型
    
    @classmethod
    async def find_all(cls):
        async with async_session_maker() as session:
            query = select(cls.model)
            result = await session.execute(query)
            # 用scalars()直接获取模型实例列表
            db_instances = result.scalars().all()
            # 转换为Pydantic响应模型列表返回
            return [cls.response_model.from_orm(instance) for instance in db_instances]

3. 在ProductService中指定响应模型：

from app.products.schemas import ProductResponse  # 导入你定义的响应模型

class ProductService(BaseService):
    model = Products
    response_model = ProductResponse

该方案最终会返回完全符合预期的扁平结构，还能自动处理空值、类型转换，避免SQLAlchemy内部属性被序列化。

方案2：修改查询语句，直接选择模型所有列

如果不想引入Pydantic，可以直接调整查询逻辑，选择模型的所有列而非整个模型对象，让mappings()返回扁平键值对：

修改BaseService的find_all方法：

@classmethod
async def find_all(cls):
    async with async_session_maker() as session:
        # 直接选择模型的所有列，而非完整实例
        query = select(*cls.model.__table__.columns)
        result = await session.execute(query)
        # 此时mappings()返回的就是扁平字典列表
        return result.mappings().all()

这种方式无需额外定义模型，适合快速调整，直接从数据库层面获取扁平字段映射。

方案3：手动展开嵌套的映射结果

如果要保留查询整个模型的方式，可以手动遍历展开嵌套内容，同时过滤SQLAlchemy内部属性：

@classmethod
async def find_all(cls):
    async with async_session_maker() as session:
        query = select(cls.model)
        result = await session.execute(query)
        # 手动遍历模型列，构建扁平字典
        return [
            {col.name: getattr(instance, col.name) for col in cls.model.__table__.columns}
            for instance in result.scalars().all()
        ]

该方案通过遍历模型列的方式避免了__dict__中包含的SQLAlchemy内部属性（如_sa_instance_state）被返回。

最终效果验证 #

无论选择哪种方案，最终的响应体都会变成你预期的扁平结构：

{
  "product_id": 1,
  "name": "Apple Pie",
  "price": 500,
  "description": "Delicious homemade apple pie",
  "stars": 4,
  "ingredients": "Apples, sugar, flour, cinnamon",
  "nutrition": "Calories: 250 per slice",
  "image_id": null
}

备注：内容来源于stack exchange，提问作者Zofiant