SQLAlchemy文本参数化查询

SQLAlchemy里写原生SQL时，直接把用户输入拼进查询字符串，等于给SQL注入攻击敞开了大门。参数化查询就是解决这个问题的标准做法——把SQL结构和数据分开传递，数据库驱动会自动处理转义和类型转换。很多开发者知道ORM的filter方法能防注入，但一遇到复杂查询要手写SQL，就习惯性地用字符串拼接，这恰恰是最危险的操作。

SQLAlchemy提供了text()构造器来创建可执行的SQL文本，它支持多种参数传递方式。最直接的是命名参数风格，用冒号加变量名作为占位符，然后通过字典传入实际值。

from sqlalchemy import text

# 命名参数方式
stmt = text("SELECT * FROM users WHERE email = :email AND status = :status")
result = session.execute(stmt, {"email": user_input_email, "status": "active"})

这种写法里，:email和:status是占位符，实际值通过字典传入。SQLAlchemy会根据底层数据库方言，自动转换成对应的参数占位符格式。PostgreSQL里变成$1、$2，MySQL里变成%s，SQLite里变成?，完全不用你操心。关键是数据库驱动在替换参数时会做正确的转义处理，恶意输入不会破坏SQL结构。

还有位置参数方式，用问号作为占位符，按顺序传入元组。不过位置参数可读性差，参数多了容易搞错顺序，生产环境建议优先用命名参数。

# 位置参数方式
stmt = text("SELECT * FROM users WHERE email = ? AND status = ?")
result = session.execute(stmt, (user_input_email, "active"))

自动推断参数类型有时候不够精确，尤其是涉及日期、JSON、枚举等特殊类型时。text()配合bindparams()可以显式声明每个参数的类型和属性，让SQLAlchemy在传递参数前做正确的类型转换。

from sqlalchemy import text, bindparam, String, Integer, Date

stmt = text("""
    SELECT * FROM orders 
    WHERE user_id = :uid 
    AND order_date >= :start_date 
    AND amount > :min_amount
""").bindparams(
    bindparam("uid", type_=Integer),
    bindparam("start_date", type_=Date),
    bindparam("min_amount", type_=Integer)
)

result = session.execute(stmt, {
    "uid": 42, 
    "start_date": datetime.date(2024, 1, 1), 
    "min_amount": 1000
})

bindparams()的另一个实用场景是设置参数默认值。有些查询条件是可选的，用default参数可以避免在字典里反复判断。

stmt = text("SELECT * FROM products WHERE category = :cat AND price < :max_price").bindparams(
    bindparam("cat", type_=String, default="electronics"),
    bindparam("max_price", type_=Integer, default=99999)
)
# 可以只传部分参数
result = session.execute(stmt, {"cat": "books"})

动态IN查询是很多开发者踩坑的地方。直接拼字符串"WHERE id IN (1,2,3)"没问题，但如果是用户输入的列表，拼字符串就危险了。SQLAlchemy的text()对IN子句有专门的处理方式，使用expanding参数标记。

stmt = text("SELECT * FROM users WHERE id IN :ids").bindparams(
    bindparam("ids", expanding=True, type_=Integer)
)

# 传入列表，SQLAlchemy会自动展开成正确的占位符
result = session.execute(stmt, {"ids": [1, 2, 3, 5, 8]})

expanding=True告诉SQLAlchemy这个参数会展开成多个占位符。传入[1,2,3]时，实际生成的SQL是WHERE id IN (:ids_1, :ids_2, :ids_3)，每个值单独绑定，完全杜绝注入风险。不同数据库对IN子句参数数量有限制，SQLAlchemy也会处理这个细节，超长列表会自动分批查询。

如果用的是较老版本的SQLAlchemy，expanding参数可能不支持，替代方案是手动生成占位符再绑定。

ids = [1, 2, 3, 5, 8]
# 生成占位符字符串
placeholders = ",".join([f":id_{i}" for i in range(len(ids))])
stmt = text(f"SELECT * FROM users WHERE id IN ({placeholders})")
# 构建参数字典
params = {f"id_{i}": val for i, val in enumerate(ids)}
result = session.execute(stmt, params)

这种方式虽然多写几行代码，但同样保证了每个值都经过参数绑定，不会直接拼接到SQL中。

参数化查询有个重要限制：占位符只能用于值，不能用于表名、列名、SQL关键字等标识符。数据库驱动在预处理阶段会解析SQL结构，标识符必须在解析前确定。如果业务需要动态表名或排序字段，参数化绑定做不到，但也不能直接拼接用户输入。

正确的做法是使用白名单校验，确保动态标识符来自你预定义的合法集合。

# 允许的表名和排序列
ALLOWED_TABLES = {"users", "orders", "products"}
ALLOWED_SORT_COLUMNS = {"id", "created_at", "updated_at", "name"}
ALLOWED_DIRECTIONS = {"ASC", "DESC"}

def get_users_by_order(table_name, sort_column, sort_direction):
    if table_name not in ALLOWED_TABLES:
        raise ValueError(f"Invalid table name: {table_name}")
    if sort_column not in ALLOWED_SORT_COLUMNS:
        raise ValueError(f"Invalid sort column: {sort_column}")
    if sort_direction.upper() not in ALLOWED_DIRECTIONS:
        raise ValueError(f"Invalid sort direction: {sort_direction}")
    
    # 经过白名单校验后，可以安全地拼接到SQL中
    stmt = text(f"""
        SELECT * FROM {table_name} 
        ORDER BY {sort_column} {sort_direction.upper()}
    """)
    return session.execute(stmt)

对于更复杂的动态查询场景，可以考虑使用SQLAlchemy Core的Table和Column对象来动态构建查询，而不是手写原始SQL。Table对象会自动引用正确的表名和列名，同时保持参数化绑定的安全性。

from sqlalchemy import Table, MetaData, select

metadata = MetaData()
users_table = Table("users", metadata, autoload_with=engine)

# 动态选择列和条件，全部参数化
query = select(users_table).where(
    users_table.c.status == bindparam("status")
).order_by(users_table.c.created_at.desc())

result = session.execute(query, {"status": "active"})

批量插入或更新时，逐条执行效率极低。SQLAlchemy支持在text()中使用多个参数集，数据库驱动会生成单条SQL配合多组参数执行。

stmt = text("""
    INSERT INTO audit_log (user_id, action, ip_address) 
    VALUES (:user_id, :action, :ip_address)
""")

# 准备多组参数
params_list = [
    {"user_id": 1, "action": "login", "ip_address": "192.168.1.10"},
    {"user_id": 2, "action": "logout", "ip_address": "192.168.1.11"},
    {"user_id": 3, "action": "update_profile", "ip_address": "192.168.1.12"},
]

# 一次调用，批量执行
session.execute(stmt, params_list)
session.commit()

这种批量参数化执行比循环单条插入快一个数量级，因为减少了数据库往返次数，同时每条数据都经过参数绑定处理。需要注意的是，不同数据库对批量参数数量有上限，超大量数据建议分批处理，每批500到1000条比较合适。

有时候不通过Session，直接用Connection对象执行SQL，参数化方式完全一样。这在需要精细控制事务或性能敏感的场景中很常见。

from sqlalchemy import create_engine

engine = create_engine("postgresql://user:pass@localhost/dbname")

with engine.connect() as conn:
    stmt = text("SELECT * FROM users WHERE email = :email")
    result = conn.execute(stmt, {"email": user_email})
    
    # 使用execution_options控制执行行为
    stmt = text("SELECT * FROM large_table WHERE status = :status")
    result = conn.execute(
        stmt.execution_options(stream_results=True, max_row_buffer=100),
        {"status": "active"}
    )

execution_options可以附加到text()语句上，控制结果流式处理、超时时间等行为，参数化绑定仍然正常工作。

参数化查询最常见的报错是参数数量不匹配或名称不对应。SQLAlchemy会抛出StatementError或DBAPIError，错误信息里通常包含具体原因。

# 错误示例：占位符和参数字典不匹配
stmt = text("SELECT * FROM users WHERE email = :email AND age > :min_age")
# 少传了min_age参数
result = session.execute(stmt, {"email": "test@example.com"})
# 报错：StatementError - 参数min_age未提供

另一个常见问题是类型转换失败。比如数据库列是Integer类型，但传入了无法转换的字符串。用bindparams()显式声明类型可以让错误在Python层面尽早暴露，而不是等到数据库返回模糊的类型错误。

调试时可以用echo=True开启SQL日志，查看实际生成的SQL和绑定的参数。

engine = create_engine("sqlite:///test.db", echo=True)
# 执行查询时会打印生成的SQL和参数

日志会显示类似SELECT * FROM users WHERE email = ? 以及绑定的参数值，方便确认参数化是否按预期工作。

参数化查询不仅是安全需要，对性能也有直接影响。数据库在处理SQL时会生成执行计划并缓存。如果SQL文本完全一致只是参数不同，数据库可以重用缓存的执行计划，避免重复解析和优化。字符串拼接会导致每个不同的参数值产生不同的SQL文本，执行计划缓存完全失效。

对于高频查询场景，这个性能差异非常显著。PostgreSQL的pg_stat_statements视图可以验证执行计划的复用情况，参数化查询的calls计数会远高于拼接方式。

不过也有例外情况。如果数据分布极不均匀，某些参数值对应的最优执行计划完全不同，数据库可能需要根据实际参数值重新规划。这时可以用SQLAlchemy的execution_options传递提示，或者使用数据库原生的计划引导功能。

# 针对PostgreSQL，可以传递查询优化提示
stmt = text("SELECT * FROM orders WHERE status = :status").execution_options(
    postgresql_include_potentially_different_plan=True
)

参数化查询是SQLAlchemy中使用原生SQL时必须掌握的核心技能。它用极小的代码代价换取了绝对的安全性，同时带来了性能提升。记住一个简单原则：任何来自外部的数据都必须通过参数绑定进入SQL，永远不要让用户输入直接接触SQL字符串。对于标识符类的动态内容，白名单校验是唯一正确的做法。