首先，本教程详细阐述了如何在 SQLAlchemy 中灵活构建动态 WHERE 查询条件。面对客户端输入的多变需求，我们通过将查询条件抽象为可迭代的列表，并结合一个通用函数进行动态应用，从而实现高度可配置的数据库查询。文章还提供了将字典形式的输入转换为 SQLAlchemy 条件表达式的实用方法，确保查询的灵活性和可维护性。
动态 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"<User(id={self.id}, name='{self.name}', email='{self.email}')>" # 定义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"<Address(id={self.id}, user_id={self.user_id}, email='{self.email_address}')>" # 数据库连接和会话设置 (仅为示例，实际应用中可能更复杂) # 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