SQLAlchemy动态查询：灵活构建WHERE条件

from sqlalchemy import select, or_
from sqlalchemy.orm import aliased
from .models import users, addresses # 假设 users 和 addresses 是 SQLAlchemy Table 对象

s = (
    select((users.c.fullname + ", " + addresses.c.email_address).label("title"))
    .where(users.c.id == addresses.c.user_id)
    .where(users.c.name.between("m", "z"))
    .where(
        or_(
            addresses.c.email_address.like("%@aol.com"),
            addresses.c.email_address.like("%@msn.com"),
        )
    )
)

from typing import List, TypeVar
from sqlalchemy import select, or_, and_
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.orm import declarative_base, relationship, Session
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey

# 假设的模型定义
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")

# 为了与原始问题中的 users.c.name 等保持一致，这里直接使用 Table 对象
# 在实际应用中，通常直接使用 User.name 等 ORM 属性
users = User.__table__
addresses = Address.__table__

# 定义泛型类型，用于确保函数类型安全
T = TypeVar("T")

def apply_filters(stmt: select[T], filters: List[ColumnElement]) -> select[T]:
    """
    将一系列过滤条件应用到 SQLAlchemy SELECT 语句上。

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

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

# 示例过滤条件列表
filters_set_1 = [
    users.c.id == addresses.c.user_id,
    users.c.name.between("m", "z")
]

filters_set_2 = [
    users.c.id == addresses.c.user_id,
    or_(
        addresses.c.email_address.like("%@aol.com"),
        addresses.c.email_address.like("%@msn.com"),
    )
]

# 构建并应用查询
# 注意：select(User) 会默认选择 User 的所有列，并返回 User 实例
# 如果需要选择特定列或进行 JOIN，需要相应调整 select() 的参数
stmt_1 = apply_filters(select(User).join(Address), filters_set_1)
stmt_2 = apply_filters(select(User).join(Address), filters_set_2)

# 打印生成的 SQL 语句（用于调试和理解）
print("--- Query 1 ---")
print(stmt_1.compile(dialect=create_engine("sqlite:///:memory:").dialect))
print("\n--- Query 2 ---")
print(stmt_2.compile(dialect=create_engine("sqlite:///:memory:").dialect))

# 实际执行查询（需要配置数据库引擎和会话）
# engine = create_engine("sqlite:///:memory:")
# Base.metadata.create_all(engine)
# SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# with SessionLocal() as session:
#     # 插入一些测试数据
#     user1 = User(name="alice", fullname="Alice Smith")
#     user2 = User(name="bob", fullname="Bob Johnson")
#     addr1 = Address(email_address="alice@aol.com", user=user1)
#     addr2 = Address(email_address="bob@msn.com", user=user2)
#     session.add_all([user1, user2, addr1, addr2])
#     session.commit()
#
#     # 执行查询
#     results_1 = session.execute(stmt_1).scalars().all()
#     print(f"Results 1: {[r.name for r in results_1]}")
#
#     results_2 = session.execute(stmt_2).scalars().all()
#     print(f"Results 2: {[r.name for r in results_2]}")

from typing import Dict, Any, List
from sqlalchemy.sql.expression import ColumnClause

def build_filters_from_dict(
    model_or_table: Base | ColumnClause,
    filter_data: Dict[str, Any]
) -> List[ColumnElement]:
    """
    根据字典数据和模型/表对象构建 SQLAlchemy 过滤条件列表。

    Args:
        model_or_table: SQLAlchemy ORM 模型类 (如 User) 或 Table 对象 (如 users)。
        filter_data: 字典，键为列名字符串，值为对应的过滤值。

    Returns:
        包含 SQLAlchemy 表达式的列表。
    Raises:
        ValueError: 如果字典中包含无效的列名。
    """
    filters = []
    # 确定是 ORM 模型还是 Table 对象
    if hasattr(model_or_table, '__table__'): # ORM Model
        table = model_or_table.__table__
    elif isinstance(model_or_table, ColumnClause): # Table object (e.g., users)
        table = model_or_table
    else:
        raise TypeError("model_or_table must be an ORM Model or a Table object.")

    for col_name, value in filter_data.items():
        if col_name not in table.c:
            raise ValueError(f"Column '{col_name}' not found in table '{table.name}'.")

        column = table.c[col_name]
        # 这里只处理简单的相等条件，可以扩展以支持更多操作符（如 > < LIKE IN）
        filters.append(column == value)
    return filters

# 示例字典输入
dynamic_filters_dict_1 = {'name': 'Alice', 'id': 1}
dynamic_filters_dict_2 = {'email_address': 'alice@aol.com'} # 假设这是针对 Address 表的

# 构建针对 User 表的查询
user_filters = build_filters_from_dict(User, dynamic_filters_dict_1)
stmt_user_dynamic = apply_filters(select(User), user_filters)
print("\n--- Dynamic Query (User) ---")
print(stmt_user_dynamic.compile(dialect=create_engine("sqlite:///:memory:").dialect))

# 构建针对 Address 表的查询 (需要注意关联表的情况)
# 如果查询 Address，则需要传入 Address 模型
address_filters = build_filters_from_dict(Address, dynamic_filters_dict_2)
stmt_address_dynamic = apply_filters(select(Address), address_filters)
print("\n--- Dynamic Query (Address) ---")
print(stmt_address_dynamic.compile(dialect=create_engine("sqlite:///:memory:").dialect))

# 结合多表查询的动态条件
# 假设我们想根据用户名称和地址邮箱进行过滤
combined_data = {'name': 'Alice', 'email_address': '%@aol.com'} # 这里的键需要区分来源

# 更复杂的场景可能需要一个映射来指明列属于哪个表
def build_combined_filters(
    user_model: Base,
    address_model: Base,
    filter_data: Dict[str, Any]
) -> List[ColumnElement]:
    filters = []
    if 'name' in filter_data:
        filters.append(user_model.name == filter_data['name'])
    if 'email_address' in filter_data:
        filters.append(address_model.email_address.like(filter_data['email_address']))
    return filters

combined_filters = build_combined_filters(User, Address, combined_data)
# 注意：如果条件涉及多表，select 语句需要包含相应的 join
stmt_combined_dynamic = apply_filters(select(User).join(Address), combined_filters)
print("\n--- Dynamic Query (Combined) ---")
print(stmt_combined_dynamic.compile(dialect=create_engine("sqlite:///:memory:").dialect))

from sqlalchemy import and_

complex_filters = [
    users.c.id == addresses.c.user_id,
    and_(users.c.name.startswith("A"), users.c.fullname.contains("Smith")), # 显式 AND
    or_(
        addresses.c.email_address.like("%@example.com"),
        addresses.c.email_address.like("%@test.org")
    )
]
stmt_complex = apply_filters(select(User).join(Address), complex_filters)
print("\n--- Complex Query ---")
print(stmt_complex.compile(dialect=create_engine("sqlite:///:memory:").dialect))