如何将SQLAlchemy查询结果转为Pydantic对象？代码正确性咨询

关于Results类的问题 #

你定义的Results泛型模型语法本身没问题，但当前代码完全没用到它，属于冗余代码。如果你的需求是用这个类封装查询结果列表，应该把字段类型改成result: list[AnyBaseModel]；但如果只是要返回Pydantic模型列表，这个类可以直接删除。

核心问题：查询结果转换错误 #

你的execute_query函数存在两个关键问题：

1. SQLAlchemy的session.execute()返回的是Result对象，不是直接可用的行数据，不能直接传给Pydantic模型构造函数
2. 函数返回类型标注是list[AnyBaseModel]，但当前代码只返回单个base_class实例，逻辑不符

修正后的完整代码 #

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, ReturningStatement
from sqlalchemy.orm import Mapped, mapped_column, Base
from pydantic import BaseModel
from pydantic.generics import GenericModel
from typing import TypeVar, Generic

class tableA(Base):
    __tablename__ = "A"

    id: Mapped[int] = mapped_column(init=False, primary_key=True)
    name: Mapped[str]

AnyBaseModel = TypeVar("AnyBaseModel", bound=BaseModel)

# 不需要封装类的话，这个Results可以直接删掉
class Results(GenericModel, Generic[AnyBaseModel]):
    """用于封装结果列表的泛型类"""
    result: list[AnyBaseModel]

class A(BaseModel):
    id: int
    name: str

    class Config:
        # 允许从ORM实例直接生成Pydantic模型（可选，简化转换）
        from_attributes = True

async def execute_query(statement: ReturningStatement, 
                        session: AsyncSession, base_class: type[AnyBaseModel]) -> list[AnyBaseModel]:
    result_rows = await session.execute(statement)
    # 提取SQLAlchemy实体实例列表
    db_items = result_rows.scalars().all()
    # 逐个转换为Pydantic模型
    return [base_class.model_validate(item) for item in db_items]

# 调用示例（注意函数名是execute_query，不是你写的execute_statement）
async def main():
    # 假设session已完成初始化
    statement = select(tableA)
    df = await execute_query(statement, session, A)

关键修正点说明 #

• 开启Pydantic模型的from_attributes配置：让模型可以直接从ORM实例（比如tableA的对象）验证生成，无需手动提取字段
• 正确处理SQLAlchemy结果：用scalars().all()提取实体实例列表，再通过model_validate逐个转换为Pydantic模型
• 修正函数返回逻辑：返回模型列表，匹配类型标注的要求

另一种转换方式（字典映射） #

如果不想开启from_attributes，可以用SQLAlchemy的mappings()方法获取字典格式的结果：

async def execute_query(statement: ReturningStatement, 
                        session: AsyncSession, base_class: type[AnyBaseModel]) -> list[AnyBaseModel]:
    result_rows = await session.execute(statement)
    db_mappings = result_rows.mappings().all()
    return [base_class(**mapping) for mapping in db_mappings]

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