SQLAlchemy 多列查询结果的对象定义保持-Python教程-PHP中文网

SQLAlchemy 多列查询结果的对象定义保持

碧海醫心

发布： 2025-09-30 17:20:14

原创

485人浏览过

sqlalchemy 多列查询结果的对象定义保持

本文介绍了在使用 SQLAlchemy 进行多表联合查询时，如何保持查询结果中每个对象的类型定义，避免类型推断为 Any。通过使用 .tuples() 方法，可以将查询结果转换为元组序列，从而方便地解包并直接使用对象，无需额外定义变量类型。

在使用 SQLAlchemy 进行数据库查询时，经常会遇到需要联合多个表进行查询的情况。 SQLAlchemy 提供了强大的 select 语句来实现这一功能。然而，当查询结果包含多个对象时，如何保持每个对象的类型定义，避免类型推断为 Any，成为了一个需要解决的问题。

问题背景

假设我们有两个模型 Item 和 Package，它们之间存在关联关系。我们想要查询同时包含 Item 和 Package 信息的记录。常见的做法是使用 select 语句进行联合查询：

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import select

# 定义数据库连接
engine = create_engine('sqlite:///:memory:')  # 使用内存数据库
Base = declarative_base()

# 定义模型
class Package(Base):
    __tablename__ = 'package'
    Package_id = Column(Integer, primary_key=True)
    name = Column(String)
    items = relationship("Item", back_populates="package")

class Item(Base):
    __tablename__ = 'item'
    Item_id = Column(Integer, primary_key=True)
    Package_id1 = Column(Integer, ForeignKey('package.Package_id'))
    description = Column(String)
    package = relationship("Package", back_populates="items")

Base.metadata.create_all(engine)

# 创建 Session
Session = sessionmaker(bind=engine)
session = Session()

# 插入一些数据
package1 = Package(name='Package 1')
package2 = Package(name='Package 2')
item1 = Item(description='Item 1', package=package1)
item2 = Item(description='Item 2', package=package1)
item3 = Item(description='Item 3', package=package2)

session.add_all([package1, package2, item1, item2, item3])
session.commit()

# 查询数据
# DB = DatabaseModel() # 假设 DatabaseModel 已经定义好 session
# stmt = select(Item, Package).join(Package, Item.Package_id1 == Package.Package_id)
# exec = DB.session.execute(stmt).all() # Sequence[Row[Tuple[Item, Package]]]

# for row in exec:
#     row #Row[Tuple[Item, Package]]

#     Item_object : Item = row[0]
#     Package_object : Package = row[1]

登录后复制

上述代码中，DB.session.execute(stmt).all() 返回的结果类型是 Sequence[Row[Tuple[Item, Package]]]。这意味着我们需要通过索引来访问 Item 和 Package 对象，并且需要手动指定类型，例如 Item_object : Item = row[0]。这种方式虽然可行，但不够优雅，并且增加了代码的冗余度。

解决方案：使用 .tuples() 方法

更简洁、更优雅的解决方案是使用 SQLAlchemy 的 .tuples() 方法。该方法可以将查询结果转换为元组序列，从而方便我们直接解包并使用对象。

标书对比王

标书对比王是一款标书查重工具，支持多份投标文件两两相互比对，重复内容高亮标记，可快速定位重复内容原文所在位置，并可导出比对报告。

查看详情

stmt = select(Item, Package).join(Package, Item.Package_id1 == Package.Package_id)
exec = session.execute(stmt).tuples().all() # Sequence[Tuple[Item, Package]]]

for _item, _package in exec:
    print(f"Item Description: {_item.description}, Package Name: {_package.name}")
    # 在这里可以直接使用 _item 和 _package 对象

登录后复制

通过在 execute 语句后添加 .tuples()，返回的结果类型变为 Sequence[Tuple[Item, Package]]]。在 for 循环中，我们可以直接使用 _item 和 _package 来访问 Item 和 Package 对象，而无需手动指定类型。

示例代码

下面是一个完整的示例代码，演示了如何使用 .tuples() 方法进行多表联合查询：

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import select

# 定义数据库连接
engine = create_engine('sqlite:///:memory:')  # 使用内存数据库
Base = declarative_base()

# 定义模型
class Package(Base):
    __tablename__ = 'package'
    Package_id = Column(Integer, primary_key=True)
    name = Column(String)
    items = relationship("Item", back_populates="package")

class Item(Base):
    __tablename__ = 'item'
    Item_id = Column(Integer, primary_key=True)
    Package_id1 = Column(Integer, ForeignKey('package.Package_id'))
    description = Column(String)
    package = relationship("Package", back_populates="items")

Base.metadata.create_all(engine)

# 创建 Session
Session = sessionmaker(bind=engine)
session = Session()

# 插入一些数据
package1 = Package(name='Package 1')
package2 = Package(name='Package 2')
item1 = Item(description='Item 1', package=package1)
item2 = Item(description='Item 2', package=package1)
item3 = Item(description='Item 3', package=package2)

session.add_all([package1, package2, item1, item2, item3])
session.commit()

# 查询数据
stmt = select(Item, Package).join(Package, Item.Package_id1 == Package.Package_id)
exec = session.execute(stmt).tuples().all()

for _item, _package in exec:
    print(f"Item Description: {_item.description}, Package Name: {_package.name}")

登录后复制

注意事项