SQLAlchemy查询结果无_asdict()方法问题及统一转字典方案

我定义了以下SQLAlchemy模型：

class Base(DeclarativeBase):
    pass

class PortfolioPerformanceTable(Base):
    __tablename__ = tn.portfolio_performance

    clientuid: Mapped[str] = mapped_column(ForeignKey(f"{tn.clients}.clientno"))
    clientno: Mapped[str] = mapped_column(ForeignKey(f"{tn.client_strategies}.clientno"))
    clientid: Mapped[str] = mapped_column(ForeignKey(f"{tn.custodian_banks}.id"), primary_key=True)

class ConsolidatedHoldingsTable(Base):
    __tablename__ = tn.consolidated_holdings

    clientuid: Mapped[str] = mapped_column(ForeignKey(f"{tn.clients}.clientno"))
    clientno: Mapped[str] = mapped_column(ForeignKey(f"{tn.client_strategies}.clientno"))
    clientid: Mapped[str] = mapped_column(ForeignKey(f"{tn.custodian_banks}.id"), primary_key=True)
    hldsdt: Mapped[str] = mapped_column(DateTime, primary_key=True)  # holdings date
    isin: Mapped[str] = mapped_column(String(12), primary_key=True)
    bbticker: Mapped[str] = mapped_column(String(100))
    bbmktsectordes: Mapped[str] = mapped_column(String(100))

并编写了两个查询函数：

def get_performances(date: str, clientids: list[str], db: Session):
    query = (
        db.query(PortfolioPerformanceTable.perfdt, PortfolioPerformanceTable.dailyrt)
        .where(PortfolioPerformanceTable.perfdt <= date)
        .where(PortfolioPerformanceTable.clientid.in_(clientids))
        .order_by(PortfolioPerformanceTable.perfdt.desc())
    )

    performances = [row._asdict() for row in query.all()]

    return performances


def get_holdings(clientids: list[str], db: Session):
    query = (
        db.query(ConsolidatedHoldingsTable)
        .filter(
            ConsolidatedHoldingsTable.hldsdt == select(func.max(ConsolidatedHoldingsTable.hldsdt))
        )
        .filter(ConsolidatedHoldingsTable.clientid.in_(clientids))
        .order_by(ConsolidatedHoldingsTable.isin.desc())
    )

    holdings = [row._asdict() for row in query.all()]

    return holdings

调用get_holdings时出现如下错误：

formance_report
    holdings = get_holdings(custodian_banks, db)
  File "/Users/mandaputra/clients_dashboard/service.py", line 39, in get_holdings
    holdings = [row._asdict() for row in query.all()]  # type: ignore
  File "/Users/mandaputra/clients_dashboard/service.py", line 39, in <listcomp>
    holdings = [row._asdict() for row in query.all()]  # type: ignore
AttributeError: 'ConsolidatedHoldingsTable' object has no attribute '_asdict'

经检查，get_performances返回的行类型为Row[Tuple[str, float]]，拥有_asdict()方法；而get_holdings返回的行类型为ConsolidatedHoldingsTable实例，无该方法。请问为何查询全模型与查询部分字段会存在此类型差异？如何实现统一将SQLAlchemy查询结果转为字典列表？

一、类型差异原因 #

• 查询指定字段（db.query(Model.col1, Model.col2)）时，SQLAlchemy返回Row对象：这是框架封装的通用行数据容器，专门提供了_asdict()方法用于快速转换为字典结构。
• 查询整个模型（db.query(Model)）时，SQLAlchemy返回模型类的实例对象：这些是ORM映射生成的实体，本质是你自定义类的实例，默认没有_asdict()方法。

二、统一转为字典列表的方法 #

方法1：给基础模型类添加通用转换方法 #

在Base类中定义to_dict方法，所有继承它的模型实例都能直接调用：

class Base(DeclarativeBase):
    def to_dict(self):
        return {col.name: getattr(self, col.name) for col in self.__table__.columns}

修改查询函数的转换逻辑：

• get_performances可保持原逻辑（Row对象用_asdict()），或统一用字典推导式（如果需要对齐逻辑）
• get_holdings改为调用模型实例的to_dict方法：

holdings = [row.to_dict() for row in query.all()]

方法2：使用SQLAlchemy内置工具转换实例 #

无需修改模型类，用SQLAlchemy的inspect工具获取实例字段信息：

from sqlalchemy import inspect

def instance_to_dict(instance):
    return {attr.key: getattr(instance, attr.key) for attr in inspect(instance).mapper.column_attrs}

在get_holdings中使用该函数：

holdings = [instance_to_dict(row) for row in query.all()]

方法3：统一查询指定字段 #

将get_holdings的查询改为指定模型的所有字段，这样返回的也是Row对象，可直接用_asdict()：

def get_holdings(clientids: list[str], db: Session):
    query = (
        db.query(
            ConsolidatedHoldingsTable.clientuid,
            ConsolidatedHoldingsTable.clientno,
            ConsolidatedHoldingsTable.clientid,
            ConsolidatedHoldingsTable.hldsdt,
            ConsolidatedHoldingsTable.isin,
            ConsolidatedHoldingsTable.bbticker,
            ConsolidatedHoldingsTable.bbmktsectordes
        )
        .filter(
            ConsolidatedHoldingsTable.hldsdt == select(func.max(ConsolidatedHoldingsTable.hldsdt))
        )
        .filter(ConsolidatedHoldingsTable.clientid.in_(clientids))
        .order_by(ConsolidatedHoldingsTable.isin.desc())
    )

    holdings = [row._asdict() for row in query.all()]

    return holdings

此方法适合字段较少的模型，字段较多时会比较繁琐。

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