SQLAlchemy 动态 WHERE 子句构建教程

Select * from users where column1 = value1;

Select * from users where column1 = value1 and column2 = value2 and column3 = value3;

import os
from typing import TypeVar, List, Dict, Any
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
from sqlalchemy.sql import select, or_

# 1. 定义 SQLAlchemy 模型
Base = declarative_base()

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

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', fullname='{self.fullname}')>"

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

    def __repr__(self):
        return f"<Address(id={self.id}, email_address='{self.email_address}')>"

# 2. 数据库连接设置 (使用内存SQLite便于演示)
engine = create_engine('sqlite:///:memory:', echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 3. 插入一些示例数据
def setup_data():
    session.add_all([
        User(name='Alice', fullname='Alice Smith', addresses=[
            Address(email_address='alice@example.com'),
            Address(email_address='alice@aol.com')
        ]),
        User(name='Bob', fullname='Bob Johnson', addresses=[
            Address(email_address='bob@msn.com')
        ]),
        User(name='Charlie', fullname='Charlie Brown', addresses=[
            Address(email_address='charlie@gmail.com')
        ])
    ])
    session.commit()

setup_data()

# 4. 定义通用的过滤器应用函数
T = TypeVar("T")

def apply_filters(st: select[T], filters: list) -> select[T]:
    """
    将一个过滤器列表应用到 SQLAlchemy Select 对象上。

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

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

# 5. 构建动态过滤器列表并应用
print("--- 示例 1: 简单过滤 ---")
# 过滤条件1：用户ID大于1，且用户名为'Bob'
filters_1 = [
    User.id > 1,
    User.name == 'Bob'
]

# 构建并执行查询
stmt_1 = apply_filters(select(User), filters_1)
result_1 = session.execute(stmt_1).scalars().all()
for user in result_1:
    print(user) # 预期输出: <User(id=2, name='Bob', fullname='Bob Johnson')>

print("\n--- 示例 2: 复杂过滤 (包含 OR 和 JOIN) ---")
# 过滤条件2：用户名为'Alice'或电子邮件地址包含'@aol.com'或'@msn.com'
# 注意：这里需要进行 JOIN 才能访问 Address 表的字段
filters_2 = [
    or_(
        User.name == 'Alice',
        Address.email_address.like('%@aol.com'),
        Address.email_address.like('%@msn.com')
    )
]

# 构建并执行查询，需要显式 JOIN
stmt_2 = apply_filters(select(User).join(User.addresses), filters_2).distinct()
result_2 = session.execute(stmt_2).scalars().all()
for user in result_2:
    print(user) # 预期输出: <User(id=1, name='Alice', fullname='Alice Smith')>, <User(id=2, name='Bob', fullname='Bob Johnson')>

print("\n--- 示例 3: 无过滤条件 ---")
# 如果没有过滤条件，列表为空
filters_3 = []
stmt_3 = apply_filters(select(User), filters_3)
result_3 = session.execute(stmt_3).scalars().all()
print(f"查询到 {len(result_3)} 个用户。") # 预期输出: 查询到 3 个用户。

# 关闭会话
session.close()