SQLAlchemy动态WHERE子句构建指南

from sqlalchemy import select, or_
from sqlalchemy.orm import declarative_base, Mapped, mapped_column
from sqlalchemy import String, Integer, ForeignKey

# 假设已定义User和Address模型
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100))
    age: Mapped[int] = mapped_column(Integer)

class Address(Base):
    __tablename__ = 'addresses'
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str] = mapped_column(String(100))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

# 静态查询示例
static_query = (
    select(User.name, Address.email_address)
    .join(Address)
    .where(User.id == Address.user_id)
    .where(User.name.between("M", "Z"))
    .where(
        or_(
            Address.email_address.like("%@aol.com"),
            Address.email_address.like("%@msn.com"),
        )
    )
)
# print(static_query) # 可以打印SQL语句查看

from typing import List, Any, TypeVar
from sqlalchemy import Select

# 定义一个类型变量，用于泛型函数，提高类型安全性
T = TypeVar("T")

def apply_filters(st: Select[T], filters: List[Any]) -> Select[T]:
    """
    将一个过滤表达式列表应用到SQLAlchemy的Select语句上。

    Args:
        st: 初始的Select语句对象。
        filters: 包含SQLAlchemy过滤表达式（如ColumnElement）的列表。

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

from sqlalchemy import and_, or_

# 示例：定义不同的过滤条件集合
filters_set_1 = [
    User.age > 25,
    User.name.like("A%"),
]

filters_set_2 = [
    User.id == Address.user_id, # 假设这是join条件，或者作为额外的过滤
    or_(
        User.age.between(20, 30),
        User.email.like("%@example.com")
    )
]

# 构建初始查询
base_query_users = select(User)
base_query_users_with_address = select(User).join(Address)

# 应用过滤器
query_1 = apply_filters(base_query_users, filters_set_1)
query_2 = apply_filters(base_query_users_with_address, filters_set_2)

# print(query_1)
# print(query_2)

from sqlalchemy import Column
from sqlalchemy.sql import expression

def get_column_from_model(model: Base, column_name_str: str) -> Column:
    """
    根据模型和字符串列名获取对应的SQLAlchemy Column对象。
    注意：这是一个简化版本，实际应用中可能需要更健壮的错误处理和反射机制。
    """
    if hasattr(model, column_name_str) and isinstance(getattr(model, column_name_str), (Column, Mapped)):
        # 对于Mapped属性，其.expression属性通常是Column对象
        col_attr = getattr(model, column_name_str)
        if isinstance(col_attr, Mapped):
            return col_attr.expression
        return col_attr
    raise ValueError(f"Column '{column_name_str}' not found or not a valid column in model '{model.__name__}'")

def build_filters_from_dict(model: Base, filter_dict: dict) -> List[Any]:
    """
    从字典构建SQLAlchemy过滤表达式列表，目前仅支持简单的相等判断。
    """
    filter_expressions = []
    for col_name, value in filter_dict.items():
        try:
            column = get_column_from_model(model, col_name)
            filter_expressions.append(column == value)
        except ValueError as e:
            print(f"Warning: {e}. Skipping filter for '{col_name}'.")
            continue
    return filter_expressions

# 客户端输入示例
client_input_1 = {"name": "Bob", "age": 30}
client_input_2 = {"email": "bob@example.com"}

# 构建过滤器列表
dynamic_filters_1 = build_filters_from_dict(User, client_input_1)
dynamic_filters_2 = build_filters_from_dict(User, client_input_2)

# 应用过滤器
query_from_dict_1 = apply_filters(select(User), dynamic_filters_1)
query_from_dict_2 = apply_filters(select(User), dynamic_filters_2)

# print(query_from_dict_1)
# print(query_from_dict_2)

def build_advanced_filters_from_dict(model: Base, filter_dict: dict) -> List[Any]:
    """
    从字典构建SQLAlchemy过滤表达式列表，支持简单的相等、LIKE、大于、小于操作。
    约定：
    - "column_name": value  -> column_name == value
    - "column_name__like": value -> column_name.like(value)
    - "column_name__gt": value -> column_name > value
    - "column_name__lt": value -> column_name < value
    - "column_name__in": list_of_values -> column_name.in_(list_of_values)
    """
    filter_expressions = []
    for key, value in filter_dict.items():
        parts = key.split('__')
        col_name = parts[0]
        operator = parts[1] if len(parts) > 1 else 'eq' # 默认相等

        try:
            column = get_column_from_model(model, col_name)
            if operator == 'eq':
                filter_expressions.append(column == value)
            elif operator == 'like':
                filter_expressions.append(column.like(value))
            elif operator == 'gt':
                filter_expressions.append(column > value)
            elif operator == 'lt':
                filter_expressions.append(column < value)
            elif operator == 'in':
                if isinstance(value, list):
                    filter_expressions.append(column.in_(value))
                else:
                    print(f"Warning: 'in' operator for '{col_name}' requires a list value. Skipping.")
            else:
                print(f"Warning: Unsupported operator '{operator}' for column '{col_name}'. Skipping.")
        except ValueError as e:
            print(f"Warning: {e}. Skipping filter for '{col_name}'.")
            continue
        except AttributeError:
            print(f"Warning: Column '{col_name}' does not support operator '{operator}'. Skipping.")
            continue
    return filter_expressions

# 客户端高级输入示例
advanced_input = {
    "name__like": "J%",
    "age__gt": 25,
    "email__in": ["john@example.com", "jane@example.com"]
}

advanced_filters = build_advanced_filters_from_dict(User, advanced_input)
advanced_query = apply_filters(select(User), advanced_filters)

# print(advanced_query)