Python SQLAlchemy - 用三个查询执行联合

在本文中，我们将看到如何在Python中使用 SQLAlchemy 执行三个查询的联合。

由于我们将在这篇文章中使用 MySQL，因此我们还将在Python中为 MySQL 安装 SQL 连接器。但是，除了 SQL 连接器之外，没有任何代码实现会随着数据库的变化而变化。

pip install pymysql

我们将使用 MySQL 中的示例 sakila 数据库。在本文中，我们将介绍 2 个示例，每个示例用于 SQLAchemy Core 和 ORM 层。在这两个示例中，我们将从 sakila 数据库中的支付表中提取“payment_id”为 1、2 或 3 的记录（作为单独的查询，然后将它们合并）。如果您没有 sakila 数据库并且想在不安装它的情况下继续阅读本文，那么使用下面提到的链接中的 SQL 脚本来创建所需的架构和付款表以及记录。

Sakila 付款表脚本

作为参考，支付表中的前十条记录如下所示——

我们在以下两个示例中查看的 SQL 查询是 -

SELECT * FROM payment WHERE payment_id = 1
UNION
SELECT * FROM payment WHERE payment_id = 2
UNION
SELECT * FROM payment WHERE payment_id = 3;

方法一：使用 SQLAlchemy Core 与三个查询进行联合：

SQLAlchemy Core 是一个以模式为中心的模型，这意味着一切都被视为数据库的一部分，即行、列、表等。在上面的示例中，我们创建了元数据对象来访问表等数据库对象。使用这个对象，我们得到了`payment`表的元数据。然后使用此元数据信息对表进行查询以获取三个查询的并集。我们首先准备将 payment_id 设置为 1、2 和 3 的单个查询。然后将这些查询作为参数传递给 sqlalchemy.union() 方法。正如我们在输出中看到的，我们得到 payment_id 为 1、2 或 3 的记录。

Syntax: sqlalchemy.sql.expression.union(*selects, **kwargs)¶

Return a UNION of multiple selectables. The returned object is an instance of CompoundSelect.

Python3

import sqlalchemy as db
  
# DEFINE THE ENGINE (CONNECTION OBJECT)
engine = db.create_engine("mysql+pymysql://root:password@localhost/sakila")
  
# CREATE THE METADATA OBJECT TO ACCESS THE TABLE
meta_data = db.MetaData(bind=engine)
db.MetaData.reflect(meta_data)
  
# GET THE `payment` TABLE FROM THE METADATA OBJECT
payment = meta_data.tables['payment']
  
# PREPARING THE REQUIRED QUERY
query_1 = db.select(payment).filter(payment.c.payment_id == 1)
query_2 = db.select(payment).filter(payment.c.payment_id == 2)
query_3 = db.select(payment).filter(payment.c.payment_id == 3)
  
query = db.union(query_1, query_2, query_3)
  
# EXTRACTING RECORDS USING THE ENGINE AND QUERY
with engine.connect() as conn:
    result = conn.execute(query).all()
  
# PRINT THE RESULTANT RECORDS
for r in result:
    print(r.payment_id, "|", r.customer_id, "|", r.rental_id, "|", r.amount)

输出：

方法2：使用SQLAlchemy ORM与三个查询进行联合：

SQLAlchemy ORM 使用以对象为中心的视图，它将模式与业务对象封装在一起。这是一个更 Pythonic 的实现，因为我们可以看到以类格式表示的表。我们已经使用这个类对象使用上面提到的 SQLAlchemy 语法来查询支付表。我们创建了单独的查询以从 payment_id 为 1、2 或 3 的表中提取记录。然后使用 union() 方法链接“sqlalchemy.orm.Query”类型的第一个查询。其他两个查询在 union() 方法中作为参数传递。一个或多个 `sqlalachemy.orm.Query` 类型的对象可以作为参数传递给此方法。

Syntax: sqlalchemy.orm.Query.union(*q)¶

Produce a UNION of this Query against one or more queries.

Python3

from sqlalchemy.orm import sessionmaker
import sqlalchemy as db
from sqlalchemy.ext.declarative import declarative_base
  
Base = declarative_base()
  
# DEFINE THE ENGINE (CONNECTIO OBJECT)
engine = db.create_engine("mysql+pymysql://root:password@localhost/sakila")
  
# CREATE THE TABLE MODEL TO USE IT FOR QUERYING
class Payment(Base):
  
    __table__ = db.Table("payment", Base.metadata, autoload_with=engine)
  
# CREATE A SESSION OBJECT TO INITIATE QUERY IN DATABASE
Session = sessionmaker(bind=engine)
session = Session()
  
# PREPARING QUERY USING SQLALCHEMY
query_1 = session.query(Payment).filter(Payment.payment_id == 1)
query_2 = session.query(Payment).filter(Payment.payment_id == 2)
query_3 = session.query(Payment).filter(Payment.payment_id == 3)
  
# EXTRACT ALL THE RECORDS BY PERFORMING UNION OF THREE QUERIES
result = query_1.union(query_2, query_3).all()
  
# PRINT THE RESULTANT RECORDS
for r in result:
    print(r.payment_id, "|", r.customer_id, "|", r.rental_id, "|", r.amount)

输出：