Python 使用 SQLAlchemy 实现动态 SQL 查询的正确方法

拾柒

发布时间：2026-10-11 | 1人浏览过

原创

针对 Web 应用中高频变动的筛选需求，本文演示如何利用 SQLAlchemy 的 ORM 特性替代硬编码 SQL。内容聚焦于条件字段的动态拼装规则，对比链式过滤、逻辑运算符集成及原始语句绑定三种方案的操作差异与适用边界。

核心实现路径

第1步：定义数据模型与初始化会话

所有动态查询均依赖预先映射的类结构与数据库会话管理器。以下为兼容主流版本的初始化基座：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建引擎（以 SQLite 为例）
engine = create_engine('sqlite:///example.db')
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)

# 绑定并实例化会话
Session = sessionmaker(bind=engine)
session = Session()

第2步：使用 filter 链式调用添加基础过滤

filter 接收模型列表达式，支持连续调用并自动合并为 AND 谓词。适合将参数字典逐项转化为查询约束：

def get_users_v1(filters):
    query = session.query(User)
    
    if 'name' in filters:
        query = query.filter(User.name == filters['name'])
        
    if 'age' in filters:
        query = query.filter(User.age >= filters['age'])
        
    return query.all()

第3步：利用 and_ 与 or_ 组合复杂业务逻辑

当需要显式指定运算优先级或混用 OR 分支时，应先收集条件对象，再统一传入逻辑函数：

from sqlalchemy import and_, or_

def get_users_v2(filters):
    query = session.query(User)
    conditions = []
    
    if 'name' in filters:
        conditions.append(User.name == filters['name'])
        
    if 'age_min' in filters:
        conditions.append(User.age >= filters['age_min'])
        
    # 按需插入 OR 逻辑
    if filters.get('status'):
        conditions.append(or_(User.status == 'active', User.status == 'pending'))
        
    if conditions:
        query = query.filter(and_(*conditions))
        
    return query.all()

第4步：通过 text 模块执行高度定制化的原始 SQL

面对 ORM 映射难以覆盖的复杂聚合或数据库方言特性，可直接拼接原生语句。必须使用命名占位符规避注入风险：

from sqlalchemy import text

def get_users_raw(filters):
    base_sql = "SELECT * FROM users WHERE 1=1"
    params = {}
    
    if 'name' in filters:
        base_sql += " AND name = :name"
        params['name'] = filters['name']
        
    if 'min_age' in filters:
        base_sql += " AND age > :min_age"
        params['min_age'] = filters['min_age']
        
    result = session.execute(text(base_sql), params)
    return result.fetchall()