在使用 SQLAlchemy 进行数据库查询时，我们经常需要选择多个表中的列，并希望能够方便地访问这些列对应的数据对象。然而，直接使用 session.execute(stmt).all() 方法可能会返回 Sequence[Row[Tuple[Item, Package]]] 这样的类型，导致在后续处理中难以直接解包并保持对象的类型信息。本文将介绍如何通过 .tuples() 方法来解决这个问题，并提供清晰的代码示例。
问题描述
假设我们有两个 SQLAlchemy 模型 Item 和 Package，并且需要通过 Item.Package_id1 和 Package.Package_id 字段进行关联查询。以下代码展示了最初的查询方式：
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:" # 使用内存数据库，方便测试
# 创建 SQLAlchemy 引擎
engine = create_engine(DATABASE_URL)
# 创建 Session 类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 定义 Base 类
Base = declarative_base()
# 定义模型
class Package(Base):
__tablename__ = "packages"
Package_id = Column(Integer, primary_key=True, index=True)
Package_name = Column(String)
items = relationship("Item", back_populates="package")
class Item(Base):
__tablename__ = "items"
Item_id = Column(Integer, primary_key=True, index=True)
Item_name = Column(String)
Package_id1 = Column(Integer, ForeignKey("packages.Package_id"))
package = relationship("Package", back_populates="items")
# 创建表
Base.metadata.create_all(bind=engine)
# 示例数据
def create_sample_data():
db = SessionLocal()
package1 = Package(Package_name="Package A")
item1 = Item(Item_name="Item 1", package=package1)
item2 = Item(Item_name="Item 2", package=package1)
db.add_all([package1, item1, item2])
db.commit()
db.close()
create_sample_data()
DB = SessionLocal()
stmt = select(Item, Package).join(Package, Item.Package_id1 == Package.Package_id)
exec = DB.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]
print(f"Item: {Item_object.Item_name}, Package: {Package_object.Package_name}")在上述代码中，exec 变量的类型为 Sequence[Row[Tuple[Item, Package]]]。这意味着我们需要通过索引来访问 Item 和 Package 对象，并且需要手动指定类型，略显繁琐。
解决方案：使用 .tuples() 方法
为了更方便地解包查询结果，我们可以使用 .tuples() 方法。该方法会将返回结果转换为 Sequence[Tuple[Item, Package]] 类型，从而可以直接使用元组解包。
以下是修改后的代码：
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:" # 使用内存数据库，方便测试
# 创建 SQLAlchemy 引擎
engine = create_engine(DATABASE_URL)
# 创建 Session 类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 定义 Base 类
Base = declarative_base()
# 定义模型
class Package(Base):
__tablename__ = "packages"
Package_id = Column(Integer, primary_key=True, index=True)
Package_name = Column(String)
items = relationship("Item", back_populates="package")
class Item(Base):
__tablename__ = "items"
Item_id = Column(Integer, primary_key=True, index=True)
Item_name = Column(String)
Package_id1 = Column(Integer, ForeignKey("packages.Package_id"))
package = relationship("Package", back_populates="items")
# 创建表
Base.metadata.create_all(bind=engine)
# 示例数据
def create_sample_data():
db = SessionLocal()
package1 = Package(Package_name="Package A")
item1 = Item(Item_name="Item 1", package=package1)
item2 = Item(Item_name="Item 2", package=package1)
db.add_all([package1, item1, item2])
db.commit()
db.close()
create_sample_data()
DB = SessionLocal()
stmt = select(Item, Package).join(Package, Item.Package_id1 == Package.Package_id)
exec = DB.execute(stmt).tuples().all() # Sequence[Tuple[Item, Package]]
for _item, _package in exec:
print(f"Item: {_item.Item_name}, Package: {_package.Package_name}")通过添加 .tuples() 方法，exec 变量的类型变为 Sequence[Tuple[Item, Package]]，我们可以在 for 循环中直接解包 Item 和 Package 对象，而无需使用索引，代码可读性更高。
总结
在 SQLAlchemy 中，使用 .tuples() 方法可以方便地处理多列选择的查询结果，避免手动指定类型和使用索引。这可以提高代码的可读性和简洁性，并减少出错的可能性。在需要同时访问多个表中的数据对象时，建议使用此方法。
