防止SQL注入在Python的SQLAlchemy中text()使用风险

在Python的SQLAlchemy中使用text()构造原生SQL语句时，最大的安全隐患就是SQL注入攻击。很多开发者以为用了ORM框架就万事大吉，但一旦你手写SQL拼接用户输入的参数，攻击者就能通过构造恶意输入篡改你的查询逻辑，甚至删库跑路。核心解决方案只有一个：永远不要用字符串拼接或格式化的方式把用户输入塞进text()里，而是使用SQLAlchemy提供的参数绑定机制，也就是冒号命名参数或者问号位置参数。下面我会把这个问题从头到尾拆解清楚，包括具体的危险写法、安全写法、底层原理以及实战中容易踩的坑。

一、text()到底是什么，为什么它有风险

SQLAlchemy的text()函数是用来把一段原生SQL字符串包装成可执行的SQL构造对象。它本身不是问题，问题在于你怎么往里面塞参数。当你写这样的代码时：

from sqlalchemy import text

# 危险写法：直接拼接用户输入
username = request.args.get('username')
query = text(f"SELECT * FROM users WHERE name = '{username}'")
result = connection.execute(query)

如果用户传入的username是 ' OR '1'='1，那最终执行的SQL就变成了 SELECT * FROM users WHERE name = '' OR '1'='1'，这会返回所有用户数据。更狠的攻击可以用 '; DROP TABLE users; -- 来直接删表。这就是SQL注入的本质：你把用户数据当成了SQL代码的一部分来执行。

二、SQLAlchemy中text()的安全使用方式——参数绑定

SQLAlchemy的text()支持两种参数绑定方式，都能有效防止注入。第一种是冒号命名参数，第二种是问号位置参数。下面分别演示。

方式一：冒号命名参数（推荐）

from sqlalchemy import text

username = request.args.get('username')
# 使用冒号参数，参数通过params字典传入
query = text("SELECT * FROM users WHERE name = :username")
result = connection.execute(query, {"username": username})

方式二：问号位置参数

from sqlalchemy import text

username = request.args.get('username')
# 使用问号占位符，参数通过元组传入
query = text("SELECT * FROM users WHERE name = ?")
result = connection.execute(query, (username,))

这两种写法的核心原理是：参数不是拼接到SQL字符串里的，而是作为独立的数据通过数据库驱动的参数化查询接口发送给数据库。数据库会先编译SQL语句的结构，再把参数作为纯数据填入，攻击者无法改变SQL的逻辑结构。这是数据库层面的防护，不是框架层面的简单过滤。

三、为什么字符串格式化和f-string绝对不能用

有些开发者会说："我先对输入做了转义处理，应该安全了吧？"答案是：不要自己做转义。原因有三个。第一，不同数据库的转义规则不一样，你写的转义函数可能只对MySQL有效，换成PostgreSQL就失效。第二，转义逻辑很容易写错，一个边界情况就可能被绕过。第三，SQLAlchemy已经帮你做了参数绑定，你为什么要重复造轮子还造得不安全？

下面是几种绝对禁止的写法：

# 禁止：f-string拼接
query = text(f"SELECT * FROM users WHERE id = {user_id}")

# 禁止：format格式化
query = text("SELECT * FROM users WHERE id = {}".format(user_id))

# 禁止：百分号格式化
query = text("SELECT * FROM users WHERE id = %s" % user_id)

这些写法无论你在前面加了多少层过滤，本质上都是把用户数据当成SQL代码的一部分，数据库收到的是一条完整的、已经被"组装"好的SQL语句，它无法区分哪部分是代码、哪部分是数据。

四、text()配合ORM的混合使用场景

实际开发中，很多时候你需要在ORM查询的基础上嵌入一段原生SQL，比如使用复杂的数据库函数、窗口函数或者特定的JSON操作。这时候text()经常和session.execute()一起出现。正确的做法是：

from sqlalchemy import text
from sqlalchemy.orm import Session

def get_active_users_with_score(session: Session, min_score: int):
    query = text("""
        SELECT u.id, u.name, u.email, 
               (SELECT COUNT(*) FROM orders o WHERE o.user_id = u.id) as order_count
        FROM users u
        WHERE u.status = :status AND u.score >= :min_score
    """)
    result = session.execute(query, {"status": "active", "min_score": min_score})
    return result.fetchall()

注意这里的 :status 和 :min_score 都是通过字典传入的绑定参数。即使 min_score 是一个整数，也不要用f-string塞进去。SQLAlchemy会自动处理类型转换，你只需要传Python的原生类型就行。

五、批量操作和IN子句的安全写法

另一个高频踩坑场景是IN子句。很多人会这样写：

# 危险写法
ids = [1, 2, 3, 4]
query = text(f"SELECT * FROM users WHERE id IN ({','.join(map(str, ids))})")

如果ids列表里的元素来自用户输入，这就有注入风险。正确的做法是用SQLAlchemy的bindparam配合展开语法：

from sqlalchemy import bindparam

ids = [1, 2, 3, 4]
# SQLAlchemy会自动生成 :id_1, :id_2, :id_3, :id_4
query = text("SELECT * FROM users WHERE id IN :ids")
result = connection.execute(query, {"ids": tuple(ids)})

或者用问号占位符的方式：

ids = [1, 2, 3, 4]
placeholders = ','.join(['?'] * len(ids))
query = text(f"SELECT * FROM users WHERE id IN ({placeholders})")
result = connection.execute(query, ids)

第二种写法看起来好像用了f-string，但注意：f-string只用来生成占位符的数量（问号的个数），实际的值还是通过参数元组传入的，数据库驱动会做参数化处理。这和直接把值拼进字符串是两回事。

六、text()在存储过程和动态SQL中的注意事项

如果你需要调用存储过程或者执行动态拼接的SQL（比如根据条件动态拼接WHERE子句），也必须遵守参数绑定原则。例如：

def build_dynamic_query(filters: dict):
    conditions = []
    params = {}
    
    for i, (key, value) in enumerate(filters.items()):
        param_name = f"param_{i}"
        conditions.append(f"{key} = :{param_name}")
        params[param_name] = value
    
    where_clause = " AND ".join(conditions) if conditions else "1=1"
    query = text(f"SELECT * FROM products WHERE {where_clause}")
    return query, params

这里的关键点是：列名（key）虽然是字符串拼接的，但参数名和参数值都是通过绑定传入的。不过这种写法有一个前提——你必须确保key（列名）是你自己代码里定义的白名单字段，而不是直接来自用户输入。如果列名也来自用户，那就需要额外做白名单校验。

七、常见误区和进阶防护建议

误区一：认为ORM的query()方法就不需要担心注入。 实际上，如果你在query()里用了text()并且拼接了参数，同样有风险。ORM只是帮你生成了大部分安全的SQL，但text()是你自己写的，责任在你。

误区二：认为用了参数绑定就可以传入任意类型的值。 参数绑定防的是SQL注入，不防业务逻辑错误。比如你把一个超长字符串绑定到一个VARCHAR(50)的字段，数据库会截断或者报错，这是另一个层面的问题。

进阶建议一：封装一个安全的执行函数。 在团队项目中，建议封装一个统一的execute函数，强制要求所有text()调用都必须通过这个函数，函数内部做参数校验和日志记录：

def safe_execute(connection, query: text, params: dict):
    if not isinstance(query, text):
        raise TypeError("query must be a text() object")
    if not isinstance(params, dict):
        raise TypeError("params must be a dictionary")
    # 记录审计日志
    logger.info(f"Executing SQL: {query.text}, params: {params}")
    return connection.execute(query, params)

进阶建议二：使用SQLAlchemy 2.0的新语法。 如果你用的是SQLAlchemy 2.0+，推荐使用新的select() API配合text()，代码更清晰，类型提示也更好：

from sqlalchemy import select, text

stmt = select(text("*")).where(text("name = :name"))
result = session.execute(stmt, {"name": username})

八、总结：记住一条铁律

在SQLAlchemy中使用text()时，永远记住一条铁律：用户输入的任何数据，都必须通过参数绑定传入，绝对不要用任何形式的字符串拼接、格式化、f-string把值塞进SQL文本里。text()本身是安全的工具，危险的是你使用它的方式。参数绑定不是可选项，是必选项。把这个习惯刻进肌肉记忆里，你的应用就能挡住绝大多数SQL注入攻击。

最后再强调一点，参数绑定防的是SQL注入，但它不能替代其他安全措施。输入验证、最小权限原则、数据库用户权限控制、WAF防护，这些都是纵深防御体系的一部分。安全从来不是单点防护，而是多层叠加。把text()的参数绑定做对，只是你安全防线的第一层，但也是最基础、最重要的一层。

防止SQL注入在Python的SQLAlchemy中text()使用风险

服务指南

帮助中心