SQLAlchemy 查询结果没有属性`_asdict()`

Question

我有这个型号。

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

这是相同的查询。我用模型来查询。但奇怪的是

_asdict

函数上没有方法

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'

我尝试检查

row

的类型，但是，奇怪的是行类型形式

get_holdings

是

ConsolidatedHoldingsTable

，而

get_performances

的行类型是

Row[Tuple[str, float]]

。

怎么会发生这种事？为什么？选择模型上的部分列与全部列有什么区别

db.query(ConsolidatedHoldingsTable)
db.query(ConsolidatedHoldingsTable.nav, ConsolidatedHoldingsTable.isin)

如何解决这个问题？我想要一致的方式将 SQLAlchemy 查询结果转换为字典列表。

Answer 1

现在我总是使用

.query(*ConsolidatedHoldingsTable.__table__.columns)

来选择所有列，因为返回的行将具有

_asdict

属性。

更好的解决方案是使用序列化库，如 @snakecharmerb 提到的 Marshmellow 或 Pydantic。