Toon de waarden in SQLAlchemy dynamische filters

Wanneer ik SQLAlchemy gebruik, gebruik ik vaak dynamische filters in mijn queries. Dit betekent dat ik begin met een lijst met enkele voorwaarden en meer voorwaarden toevoeg die afhankelijk zijn van andere variabelen.
Hier is een query met een statisch filter:

# 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)

We kunnen dit veranderen met een dynamisch filter:

# 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)

Het filter is een lijst van BinaryExpression objecten. Het is eenvoudig om ze af te drukken:

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

En het resultaat is bijvoorbeeld:

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

Dat is erg mooi, maar niet erg handig voor debuggen omdat de waarden niet worden getoond.

Ik heb een kleine helperfunctie 'dump_filter_items' gemaakt die de filter_items inclusief hun waarden dumpt. Deze functie produceert een meer detailed dump van de filter_items en bevat de waarden:

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)

Dit is erg eenvoudig, er is veel meer informatie in het object BinaryExpression.

Hier is de code als je het zelf wilt proberen.

# 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)

Samenvatting

Ik wilde de waarden van SQLAlchemy dynamische filters controleren in mijn code. Ik heb een kleine helperfunctie gemaakt die het debuggen gemakkelijker maakt.

Links / credits

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