FastAPI+SQLAlchemy返回带列名字典列表报错求助

跟着FastAPI+SQLAlchemy视频教程写代码时遇到两个问题：

1. 直接用教程里的代码运行报错：Cannot convert dictionary update sequence element #0 to a sequence
2. 尝试两种临时处理方式都达不到预期：
   • result.scalars().all()只能返回id数值，拿不到其他字段数据
   • 用chain处理后能拿到数据，但返回的是无列名的纯值列表，用户无法清晰理解对应字段

想要实现和教程一样的效果：返回包含列名的字典列表。

教程代码：

@router.get("/")
async def get_specific_operations(operation_type: str, session: AsyncSession = Depends(get_async_session)):
    query = select(operation).where(operation.c.type == operation_type)
    result = await session.execute(query)
    return result.all()

尝试的chain代码：

@router.get("/")
async def get_specific_operations(operation_type: str, session: AsyncSession = Depends(get_async_session)):
    query = select(operation).where(operation.c.type == operation_type)
    result = await session.execute(query)
    result = list(chain(*result))
    return result

当前返回（chain处理后） #

[
 4,
 "name",
 25,
 7,
 "name2",
 31
]

期望返回 #

[
    {
        "id": 4,
        "username": "name",
        "age": 25
    },
    {
        "id": 7,
        "username": "name2",
        "age":31
    }
]

问题根源是SQLAlchemy版本差异：旧版本中result.all()直接返回可序列化的字典/结构，但新版本返回的是Row对象列表，直接返回会触发FastAPI序列化错误。下面两种方法可以解决：

方法1：将Row对象转为字典（用._mapping属性） #

修改代码，遍历结果把每个Row对象转成字典：

@router.get("/")
async def get_specific_operations(operation_type: str, session: AsyncSession = Depends(get_async_session)):
    query = select(operation).where(operation.c.type == operation_type)
    result = await session.execute(query)
    return [row._mapping for row in result.all()]

方法2：用官方提供的mappings()方法（更规范） #

SQLAlchemy提供了直接返回字典格式结果集的方法，无需手动转换：

@router.get("/")
async def get_specific_operations(operation_type: str, session: AsyncSession = Depends(get_async_session)):
    query = select(operation).where(operation.c.type == operation_type)
    result = await session.execute(query)
    return result.mappings().all()

补充说明 #

• scalars().all()只提取查询结果的第一列（这里对应id字段），所以拿不到其他字段
• chain处理是把Row对象拆成了纯值序列，自然丢失了列名映射关系

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