动态 WHERE 条件的需求与挑战
在开发数据库驱动的应用程序时，经常需要根据用户输入或业务逻辑的变化来动态调整查询条件。例如，一个数据查询接口可能接收一个包含多个过滤字段的字典，而这些字段的数量和组合是不确定的。
考虑以下两种典型的动态查询场景：
- Select * from users where column1 = value1
- Select * from users where column1 = value1 and column2 = value2 and column3 = value3
在 SQLAlchemy 中，静态的 where 子句链式调用非常直观，如 select(...).where(condition1).where(condition2)。然而，当条件数量和具体内容需要在运行时根据输入（例如 d_1 = {'column1': 'value1'} 或 d_2 = {'column1': value1, 'column2': value2, 'column3': value3}）动态增减时，这种静态模式就显得力不从心。我们需要一种机制来灵活地构建和应用这些条件。
核心策略：条件列表与迭代应用
解决动态 WHERE 条件问题的核心思想是：将所有待应用的条件收集到一个列表中，然后遍历这个列表，逐一将条件应用到 select 对象上。这种方法将条件的生成与条件的实际应用解耦，大大增强了查询的灵活性。
为了演示这一策略，我们首先定义一些 SQLAlchemy 模型或表结构。这里我们使用声明式基类（Declarative Base）来创建 User 和 Address 模型。
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, or_
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
from sqlalchemy import select
from typing import TypeVar, List
# 声明式基类
Base = declarative_base()
# 定义User模型
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
addresses = relationship("Address", back_populates="user")
def __repr__(self):
return f""
# 定义Address模型
class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
email_address = Column(String)
user = relationship("User", back_populates="addresses")
def __repr__(self):
return f""
# 数据库连接和会话设置 (仅为示例，实际应用中可能更复杂)
# engine = create_engine('sqlite:///:memory:')
# Base.metadata.create_all(engine)
# Session = sessionmaker(bind=engine)
# session = Session() 接下来，我们实现一个通用函数 apply_filters，它接受一个 select 对象和一个条件列表，并依次将列表中的每个条件应用到 select 对象上。
# 定义泛型类型，以支持类型提示
T = TypeVar("T")
def apply_filters(st: select[T], filters: List) -> select[T]:
"""
将一个条件列表动态应用到 SQLAlchemy 的 select 对象上。
Args:
st: 初始的 select 对象。
filters: 包含 SQLAlchemy 条件表达式的列表。
Returns:
应用了所有条件的 select 对象。
"""
for flt in filters:
st = st.where(flt)
return st现在，我们可以通过构建不同的条件列表来生成动态查询：
# 示例：构建不同的条件列表
# 条件列表1：筛选用户ID和名称范围
filters_1 = [
User.id == Address.user_id, # 假设我们需要联接
User.name.between("A", "M")
]
# 条件列表2：筛选用户ID和邮箱地址
filters_2 = [
User.id == Address.user_id,
or_(
Address.email_address.like("%@aol.com"),
Address.email_address.like("%@msn.com"),
)
]
# 应用条件列表生成查询
# 注意：这里为了简化，假设User和Address是直接可用的，
# 实际中可能需要通过 join 来关联
st_1 = apply_filters(select(User, Address).join(Address), filters_1)
st_2 = apply_filters(select(User, Address).join(Address), filters_2)
# 打印生成的SQL语句 (用于验证，需要一个已配置的 engine)
# print("查询1的SQL:", st_1.compile(dialect=engine.dialect))
# print("查询2的SQL:", st_2.compile 
