本文介绍了在使用 SQLAlchemy 进行多表查询时，如何保持返回对象的类型定义，避免类型推断为 "Any"。通过使用 .tuples() 方法，可以有效地将查询结果转换为元组，从而方便地解包并保持对象类型的明确性，提升代码的可读性和可维护性。
在使用 SQLAlchemy 进行数据库查询时，经常需要从多个表中选择数据。默认情况下，session.execute(stmt).all() 返回的结果类型是 Sequence[Row[Tuple[...]]]，其中 Row 对象封装了查询结果，这在某些情况下可能会导致类型推断问题，尤其是在需要明确对象类型时。本文将探讨如何解决这个问题，并提供一种更简洁明了的方法来保持对象定义。
问题描述
当使用 SQLAlchemy 进行多表联合查询时，例如：
DB = DatabaseModel()
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]直接从 Row 对象中提取数据时，需要通过索引访问，并且需要手动指定类型，例如 Item_object : Item = row[0]。 这种方式略显繁琐，且不够直观。
解决方案：使用 .tuples() 方法
SQLAlchemy 提供了 .tuples() 方法，可以将查询结果直接转换为元组形式，从而避免了 Row 对象的封装。修改后的代码如下：
DB = DatabaseModel()
stmt = select(Item, Package).join(Package, Item.Package_id1 == Package.Package_id)
exec = DB.session.execute(stmt).tuples().all() # Sequence[Tuple[Item, Package]]]
for _item, _package in exec:
# 现在可以直接使用 _item 和 _package，它们分别是 Item 和 Package 的实例
print(_item.name)
print(_package.description)通过在 execute() 语句后添加 .tuples()，返回的结果类型变为 Sequence[Tuple[Item, Package]]]。 这使得在循环中可以直接解包元组，并将查询结果赋值给对应的变量，无需手动指定类型，代码更加简洁易懂。
示例代码
以下是一个完整的示例代码，展示了如何使用 .tuples() 方法进行多表查询并保持对象定义：
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, select
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
# 定义数据库连接
DATABASE_URL = "sqlite:///:memory:" # 使用内存数据库，方便演示
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 定义模型
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String)
package_id = Column(Integer, ForeignKey("packages.id"))
package = relationship("Package", back_populates="items")
class Package(Base):
__tablename__ = "packages"
id = Column(Integer, primary_key=True, index=True)
description = Column(String)
items = relationship("Item", back_populates="package")
# 创建数据库表
Base.metadata.create_all(bind=engine)
# 创建一些示例数据
def create_sample_data():
db = SessionLocal()
package1 = Package(description="Package 1")
package2 = Package(description="Package 2")
item1 = Item(name="Item 1", package=package1)
item2 = Item(name="Item 2", package=package1)
item3 = Item(name="Item 3", package=package2)
db.add_all([package1, package2, item1, item2, item3])
db.commit()
db.close()
create_sample_data()
# 查询数据
def query_data():
db = SessionLocal()
stmt = select(Item, Package).join(Package, Item.package_id == Package.id)
results = db.execute(stmt).tuples().all()
for item, package in results:
print(f"Item Name: {item.name}, Package Description: {package.description}")
db.close()
query_data()注意事项
- .tuples() 方法返回的是元组，因此需要确保解包时的变量数量与查询结果的列数一致。
- 如果需要访问原始的 Row 对象，则不应使用 .tuples() 方法。
- 此方法适用于 SQLAlchemy Core 和 SQLAlchemy ORM。
总结
通过使用 SQLAlchemy 的 .tuples() 方法，可以更方便地处理多表查询结果，并保持对象定义的明确性。这种方法简化了代码，提高了可读性，并避免了手动类型转换的需要。在需要明确对象类型和简化代码的场景下，推荐使用此方法。
