SQLAlchemy 动态 WHERE 子句构建指南

from sqlalchemy import select, or_, create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker, relationship

# 假设的ORM模型定义
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="addresses")

# 示例查询：静态WHERE子句
static_query = (
    select(User.fullname, Address.email_address)
    .join(Address)
    .where(User.id == Address.user_id)
    .where(User.name.between("m", "z"))
    .where(
        or_(
            Address.email_address.like("%@aol.com"),
            Address.email_address.like("%@msn.com"),
        )
    )
)

# 客户端可能提供的动态过滤条件
d_1 = {"column1": "value1"}
d_2 = {"column1": "value1", "column2": "value2", "column3": "value3"}

from typing import TypeVar, List, Any
from sqlalchemy.sql import Select, ColumnElement

# 定义一个类型变量，用于泛型函数
T = TypeVar("T")

def apply_filters(statement: Select[T], filters: List[ColumnElement[Any]]) -> Select[T]:
    """
    将一系列SQLAlchemy过滤条件动态应用到Select语句上。

    Args:
        statement: 初始的SQLAlchemy Select语句对象。
        filters: 包含SQLAlchemy表达式的列表，每个表达式代表一个WHERE条件。

    Returns:
        应用了所有过滤条件后的Select语句对象。
    """
    for flt in filters:
        statement = statement.where(flt)
    return statement

# 假设我们已经有了User和Address模型定义
# ... (User和Address模型定义如上文所示)

# 初始化数据库和会话（仅为演示目的）
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 插入一些示例数据
user1 = User(name='Alice', fullname='Alice Smith')
user2 = User(name='Bob', fullname='Bob Johnson')
user3 = User(name='Charlie', fullname='Charlie Brown')
user4 = User(name='Michael', fullname='Michael Scott')
user5 = User(name='Zoe', fullname='Zoe Miller')

address1 = Address(user=user1, email_address='alice@example.com')
address2 = Address(user=user2, email_address='bob@aol.com')
address3 = Address(user=user3, email_address='charlie@msn.com')
address4 = Address(user=user4, email_address='michael@yahoo.com')
address5 = Address(user=user5, email_address='zoe@aol.com')

session.add_all([user1, user2, user3, user4, user5, address1, address2, address3, address4, address5])
session.commit()

# 初始查询，选择User模型的所有列
base_query = select(User)

# 过滤条件集合 1：查找名字在 'm' 到 'z' 之间的用户
filters_1 = [
    User.name.between("m", "z")
]

# 过滤条件集合 2：查找邮件地址为 '@aol.com' 或 '@msn.com' 的用户
# 注意：这里需要先join Address表才能访问Address.email_address
filters_2 = [
    or_(
        Address.email_address.like("%@aol.com"),
        Address.email_address.like("%@msn.com"),
    )
]

# 过滤条件集合 3：结合多个条件，例如名字和邮件后缀
filters_3 = [
    User.name.between("m", "z"),
    or_(
        Address.email_address.like("%@aol.com"),
        Address.email_address.like("%@msn.com"),
    )
]

# 应用过滤条件
print("--- 查询 1：名字在 'm' 到 'z' 之间 ---")
# 注意：如果条件涉及关联表，需要在base_query中先join
query_1 = apply_filters(base_query, filters_1)
for user in session.scalars(query_1).all():
    print(f"User ID: {user.id}, Name: {user.name}, Fullname: {user.fullname}")

print("\n--- 查询 2：邮件地址为 '@aol.com' 或 '@msn.com' ---")
# 这里的base_query需要包含join操作，以便访问Address表的列
query_2_base = select(User).join(Address)
query_2 = apply_filters(query_2_base, filters_2)
for user in session.scalars(query_2).all():
    print(f"User ID: {user.id}, Name: {user.name}, Email: {[a.email_address for a in user.addresses]}")

print("\n--- 查询 3：名字在 'm' 到 'z' 之间 且 邮件地址为 '@aol.com' 或 '@msn.com' ---")
query_3_base = select(User).join(Address)
query_3 = apply_filters(query_3_base, filters_3)
for user in session.scalars(query_3).all():
    print(f"User ID: {user.id}, Name: {user.name}, Email: {[a.email_address for a in user.addresses]}")

session.close()

--- 查询 1：名字在 'm' 到 'z' 之间 ---
User ID: 4, Name: Michael, Fullname: Michael Scott
User ID: 5, Name: Zoe, Fullname: Zoe Miller

--- 查询 2：邮件地址为 '@aol.com' 或 '@msn.com' ---
User ID: 2, Name: Bob, Email: ['bob@aol.com']
User ID: 3, Name: Charlie, Email: ['charlie@msn.com']
User ID: 5, Name: Zoe, Email: ['zoe@aol.com']

--- 查询 3：名字在 'm' 到 'z' 之间 且 邮件地址为 '@aol.com' 或 '@msn.com' ---
User ID: 5, Name: Zoe, Email: ['zoe@aol.com']