Python SQLAlchemy: realización de unión con tres consultas

En este artículo, veremos cómo realizar una unión de tres consultas usando SQLAlchemy en Python.

Como vamos a usar MySQL en esta publicación, también instalaremos un conector SQL para MySQL en Python. Sin embargo, ninguna de las implementaciones de código cambia con el cambio en la base de datos, excepto los conectores SQL.

pip install pymysql

Usaremos la base de datos sakila de muestra de MySQL. En este artículo, cubriremos 2 ejemplos, uno para las capas SQLAchemy Core y ORM. En ambos ejemplos, extraeremos los registros de la tabla de pagos dentro de la base de datos de sakila que tienen un `pago_id` de 1, 2 o 3 (como consultas individuales y luego tomaremos la unión de ellos). Si no tiene la base de datos sakila y desea seguir este artículo sin instalarlo, use el script SQL presente en el enlace que se menciona a continuación para crear el esquema requerido y la tabla de pagos junto con los registros.

Guión de tabla de pagos de Sakila

Como referencia, los primeros diez registros en la tabla de pagos se ven así:

La consulta SQL que estamos viendo en los dos ejemplos siguientes es:

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

Método 1: usar SQLAlchemy Core para realizar la unión con tres consultas:

SQLAlchemy Core es un modelo centrado en el esquema que significa que todo se trata como parte de la base de datos, es decir, filas, columnas, tablas, etc. En el ejemplo anterior, hemos creado el objeto de metadatos para acceder a los objetos de la base de datos como la tabla. Usando este objeto obtenemos los metadatos de la tabla `pago`. Esta información de metadatos luego se usa para consultar la tabla para obtener la unión de las tres consultas. Primero preparamos consultas individuales con payment_id como 1, 2 y 3. Luego, estas consultas se pasan como parámetros en el método sqlalchemy.union(). Como podemos ver en la salida, obtenemos los registros que tienen pay_id como 1, 2 o 3.

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

Devuelve una UNIÓN de múltiples seleccionables. El objeto devuelto es una instancia de CompoundSelect.

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)

Producción:

Método 2: usar SQLAlchemy ORM para realizar la unión con tres consultas:

SQLAlchemy ORM utiliza una vista centrada en objetos que encapsula el esquema con objetos comerciales. Es una implementación más pythonica ya que podemos ver las tablas representadas en el formato de clase. Hemos usado este objeto de clase para consultar la tabla de pagos usando la sintaxis de SQLAlchemy mencionada anteriormente. Creamos consultas individuales para extraer registros de la tabla que tienen id_pago como 1, 2 o 3. Luego, la primera consulta, que es del tipo `sqlalchemy.orm.Query`, se enstring mediante el método union(). Las otras dos consultas se pasan como parámetros en el método union(). Uno o más objetos de tipo `sqlalachemy.orm.Query` se pueden pasar como parámetros a este método.

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

Producir una UNIÓN de esta Consulta contra una o más consultas.

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)

Producción: