SQLAlchemy动态查询条件构建教程

动态 WHERE 条件的需求与挑战

在开发数据库驱动的应用程序时，经常需要根据用户输入或业务逻辑的变化来动态调整查询条件。例如，一个数据查询接口可能接收一个包含多个过滤字段的字典，而这些字段的数量和组合是不确定的。

考虑以下两种典型的动态查询场景：

1. Select * from users where column1 = value1
2. 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