如何将 Sqlalchemy ORM 查询结果转换为包含关系的单个联接表？

Posted 2023-04-12

【中文标题】如何将 Sqlalchemy ORM 查询结果转换为包含关系的单个联接表？

【英文标题】：How can I convert Sqlalchemy ORM query results into a single joined table including relationships?

【发布时间】：2019-07-08 18:52:29

【问题描述】：

我正在为 Web 应用程序数据库使用现有的 Python API，并希望为用户提供查询结果的格式选项。在某些情况下，JSON 是合适的，而在其他情况下是 CSV。 API 是围绕 SQLAlchemy 声明式 ORM 构建的，系统中的查询通常返回 ORM 实例或实例列表及其关系。将其转换为 JSON 很简单，但是我无法检索类似表格的视图以将结果转换为 CSV。使用 SQLAlchemy 核心操作，或使用纯 SQL，很容易取回连接的表视图。我知道在幕后 SQLAlchemy 正在构建必要的 SQL 语句并连接来执行我的要求，但它隐藏在映射器后面。

鉴于我正在使用的 API 已经设置为使用 ORM 类以声明方式查询所有内容，我如何才能在 SQLAlchemy 执行其魔术将所有内容映射回对象之前取回数据库返回的行？

user_addresses = Table("user_addresses", Base.metadata,
    Column("user_id", Integer, ForeignKey("user.id")),
    Column("address_id", Integer, ForeignKey("address.id"))
)

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    addresses = relationship("Address", secondary=user_addresses, back_populates="users")


class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True, autoincrement=True)
    number = Column(Integer)
    street = Column(String)
    city = Column(String)
    state = Column(String)
    zip = Column(Integer)
    users = relationship("User", secondary=user_addresses, back_populates="addresses")


## configure session, engine, etc.
# get all relationships loaded together
statement = session.query(User).options(joinedload("*"))
# normal orm classes
results = statement.all()
# I want something like
flat_results = statement.table()

我认为像上面这样的东西必须直接与预先映射的表进行交互，但我不确定如何在不为后端重写查询 API 的情况下返回表。

【参考方案1】：

对我自己的问题给出一个可能的答案，所以如果您有更好的答案，请随时回复。

到目前为止，我发现实现这一目标的最简单方法是提取将由 Session 执行的实际 SQL 语句，并使用 execute 方法直接执行它：

statement = session.query(User).options(joinedload("*"))
results = session.execute(str(statement))
colnames = results.keys()
rows = results.fetchall()

这样，您可以利用 SQLAlchemy 的所有自动连接，并且仍然可以获取数据的完整表视图。

编辑： 正如@IljaEverilä 所指出的，最好调用Query.statement 来获取具有所有正确参数替换和后端特定功能的最终SQL 命令：

statement = session.query(User).options(joinedload("*")).statement
results = session.execute(statement)
colnames = results.keys()
rows = results.fetchall()

【讨论】：

session.execute(str(statement)) 如果您使用任何参数或特定方言的 SQL，将会崩溃。相反，您可以从 Query.statement 访问底层核心语句并传递它。