当前位置：首页 > 文章列表 > 文章 > python教程 > SQLAlchemy查询多列并保留对象结构

SQLAlchemy查询多列并保留对象结构

2025-10-06 11:48:28 0浏览收藏

推广推荐

支持 PC / 移动端，安全直达

编程并不是一个机械性的工作，而是需要有思考，有创新的工作，语法是固定的，但解决问题的思路则是依靠人的思维，这就需要我们坚持学习和更新自己的知识。今天golang学习网就整理分享《使用 SQLAlchemy 查询多列并保持对象定义》，文章讲解的知识点主要包括，如果你对文章方面的知识点感兴趣，就不要错过golang学习网，在这可以对大家的知识积累有所帮助，助力开发能力的提升。

使用 SQLAlchemy 查询时保持对象定义并进行多列选择

本文介绍了在使用 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()