Mostrar los valores en filtros dinámicos SQLAlchemy

Cuando utilizo SQLAlchemy, suelo utilizar filtros dinámicos en mis consultas. Esto significa que empiezo con una lista con algunas condiciones y añado más condiciones que dependen de otras variables.
Aquí tienes una consulta con un filtro estático:

# query with static filter

product_colors = ['white']

stmt = sa.select(Product).\
    where(sa.and(
        Product.category.in_(my_categories),
        Product.color.in_(product_colors),
        Product.price < 400,
        ...
    )).\
    order_by(Product.price)

Podemos cambiarlo utilizando un filtro dinámico:

# query with dynamic filter

product_colors = ['white']

# start with:
filter_items = [
    Product.category.in_(my_categories)
]

# add somewhere else:
filter_items.extend([
    Product.color.in_(product_colors),
    Product.price < 400,
    ...
])
    
# construct query
filter_tuple = tuple(filter_items)
stmt = sa.select(Product).\
    where(sa.and_(*filter_tuple)).\
    )).\
    order_by(Product.price)

El filtro es una lista de objetos BinaryExpression. Es fácil imprimirlos:

print(f'filter_items:')
for i, filter_item in enumerate(filter_items):
    print(f'[{i}] {filter_item}')

Y el resultado es, por ejemplo:

filter_items:
[0] product.customer_id = customer.id
[1] product.category = :category_1
[2] product.color IN (__[POSTCOMPILE_color_1])
[3] product.price IS NOT NULL
[4] product.price < :price_1

Eso está muy bien, pero no es muy útil para depurar porque no se muestran los valores.

He creado una pequeña función de ayuda 'dump_filter_items' que vuelca los filter_items incluyendo sus valores. Esta función produce un volcado más detailed de los filter_items e incluye los valores:

filter_items with values:
[0] product.customer_id = customer.id
[0] - left: product.customer_id
[0] - operator: <built-in function eq>
[0] - right: customer.id
[1] product.category = :category_1
[1] - left: product.category
[1] - operator: <built-in function eq>
[1] - right: shirt (value)
[2] product.color IN (__[POSTCOMPILE_color_1])
[2] - left: product.color
[2] - operator: <function in_op at 0x7f88efce5260>
[2] - right: ['white'] (value)
[3] product.price IS NOT NULL
[3] - left: product.price
[3] - operator: <function is_not at 0x7f88efce4cc0>
[3] - right: NULL
[4] product.price < :price_1
[4] - left: product.price
[4] - operator: <built-in function lt>
[4] - right: 400 (value)

Esto es muy básico, hay mucha más información en el objeto BinaryExpression.

Aquí tienes el código por si quieres probar tú mismo.

# customer_product.py
import sys

import sqlalchemy as sa
from sqlalchemy.orm import declarative_base, sessionmaker, relationship

from sqlalchemy.engine import Engine
from sqlalchemy import event
from sqlite3 import Connection as SQLite3Connection

# ENABLE FOREIGN KEYS
@event.listens_for(Engine, "connect")
def _set_sqlite_pragma(dbapi_connection, connection_record):
    if isinstance(dbapi_connection, SQLite3Connection):
        print('turning on foreign keys ...')
        cursor = dbapi_connection.cursor()
        cursor.execute('PRAGMA foreign_keys=ON;')
        cursor.close()

Base = declarative_base()

class Customer(Base):
    __tablename__ = 'customer'

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String(100), nullable=False)
    city = sa.Column(sa.String(100), nullable=False)
    age = sa.Column(sa.Integer, nullable=False)

    # relationship: orders
    products = relationship(
        'Product',
        back_populates='customer',
    )

    def __repr__(self):
        return f'<Customer: id = {self.id}, name = {self.name}, city = {self.city}, age = {self.age}>'


class Product(Base):
    __tablename__ = 'product'

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String(100), nullable=False)
    category = sa.Column(sa.String(100), nullable=False)
    color = sa.Column(sa.String(100), nullable=False)
    price = sa.Column(sa.Integer, nullable=False)

    # relationship: customer
    customer_id = sa.Column(sa.Integer, sa.ForeignKey('customer.id'), nullable=False, index=True)
    customer = relationship(
        'Customer',
        back_populates='products',
    )

    def __repr__(self):
        return f'<Product: id = {self.id}, name = {self.name}, color = {self.color}, price = {self.price}>'

# get engine and create all
engine_echo = True
engine = sa.create_engine('sqlite:///:memory:', echo=engine_echo)
Base.metadata.drop_all(engine, checkfirst=True)
Base.metadata.create_all(engine)

print('create session')
Session = sessionmaker(bind=engine)
session = Session()

print('add some customers ...')
customers = [
    Customer(name='John', city='Houston', age='45'), 
    Customer(name='Jane', city='red', age='44'), 
    Customer(name='Bob',  city='blue', age='60'), 
    Customer(name='Alice', city='white', age='55'), 
]
session.add_all(customers)

print('add some products ...')
products = [
    Product(name='Shirt1', category='shirt', color='white', price='100'), 
    Product(name='Dress1', category='dress', color='red', price='200'), 
    Product(name='Socks1', category='socks', color='blue', price='300'), 
    Product(name='Shirt2', category='shirt', color='white', price='300'), 
]
session.add_all(products)

print('add some products to some customers ...')
customers[0].products = [products[0], products[1]]
customers[1].products = [products[1], products[2]]
customers[2].products = [products[1], products[2], products[3]]

# stuff
session.commit()
session.flush()

product_colors = ['white']

# query1: select all customers with products: white shirts with price below 400, sort by highest price
stmt = sa.select(Customer, Product).\
    where(sa.and_(
        Product.customer_id == Customer.id,
        Product.category == 'shirt',
        Product.color.in_(product_colors),
        Product.price != None,
        Product.price < 400,
    )).\
    order_by(Product.price.desc())
customer_product = session.execute(stmt).all()

print(f'query1 result:')
for customer, product in customer_product:
    print(f'{customer}: {product}')

# now use filter_items

# start with:
filter_items = [
    Product.customer_id == Customer.id,
    Product.category == 'shirt',
]

# add somewhere else:
filter_items.extend([
    Product.color.in_(product_colors),
    Product.price != None,
    Product.price < 400,
])

print(f'filter_items:')
for i, filter_item in enumerate(filter_items):
    print(f'[{i}] {filter_item}')

filter_tuple = tuple(filter_items)
stmt = sa.select(Customer, Product).\
    where(sa.and_(*filter_tuple)).\
    order_by(Product.price.desc())
customer_product = session.execute(stmt).all()

print(f'query2 result:')
for customer, product in customer_product:
    print(f'{customer}: {product}')

# at a certain moment you want to see the values in the filter_items
# every filter_item is a sqlalchemy.sql.elements.BinaryExpression object
def dump_filter_items(filter_items):
    for i, filter_item in enumerate(filter_items):
        print(f'[{i}] {filter_item}')
        #print(f'[{i}] dict = {filter_item.__dict__}')
        if isinstance(filter_item, sa.sql.elements.BinaryExpression):
            print(f'[{i}] - left: {filter_item.left}')
            print(f'[{i}] - operator: {filter_item.operator}')
            if not hasattr(filter_item, 'right'):
                print(f'[{i}] - filter_item.right not present')
            else:
                filter_item_right = getattr(filter_item, 'right')
                if not hasattr(filter_item_right, 'value'):
                    print(f'[{i}] - right: {filter_item_right}')
                else:
                    value = getattr(filter_item_right, 'value')
                    print(f'[{i}] - right: {filter_item.right.value} (value)')
        else:
            print(f'[{i}] ?')

print(f'filter_items with values:')
dump_filter_items(filter_items)

Resumen

Quería comprobar los valores de los filtros dinámicos SQLAlchemy en mi código. He creado una pequeña función de ayuda que me hace la vida de depuración más fácil.

Enlaces / créditos

Get filtered values from SQL Alchemy ORM Query
https://stackoverflow.com/questions/47478837/get-filtered-values-from-sql-alchemy-orm-query

SQLAlchemy - Column Elements and Expressions
https://docs.sqlalchemy.org/en/20/core/sqlelement.html