本文介绍了在使用 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() 方法。该方法可以将查询结果转换为元组序列，从而方便我们直接解包并使用对象。
基于ThinkPhp6+ swoole4+uniapp 开发的一套CRMEB新零售多商户商城系统。如果不会搭建请到 查看搭建说明系统环境推荐 使用 宝塔配置环境centos PHP7.3 mysql5.6新增功能： 01·新增支持销售虚拟产品自动发货 02.支持销售链接与卡密可导入导出 03.自定义后台路径对后台进行保护 04.新增支持商家缴纳保证金功能 05·违法或侵权商品一键举报功能 06·仲
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}")注意事项
- .tuples() 方法只能用于返回多个对象的查询。如果查询只返回一个对象，则不需要使用该方法。
- 使用 .tuples() 方法后，查询结果将不再是 Row 对象，而是元组。因此，不能再使用索引来访问对象。
总结
通过使用 SQLAlchemy 的 .tuples() 方法，我们可以更方便地进行多表联合查询，并保持查询结果中每个对象的类型定义。这种方法不仅简化了代码，还提高了代码的可读性。在实际开发中，可以根据具体情况选择合适的查询方式，以达到最佳的效果。
