我正在使用 SQLAlchemy 连接到 Python 中的不同数据库，但不支持 ORM，因为由于多种原因，这无法实现。
我主要使用类似的东西构建一个复杂的 SQL 查询
sql += "AND fieldname = '%s'" % myvar
在我的情况下不是 SQL 注入问题，因为数据始终来自受信任的来源，但即使来源受信任，它也可能包含可能破坏查询的字符，例如 '、% 或 _ .
主要是我需要对它们进行转义，我想知道是否有一个已经存在的转义函数可以重复使用。
【问题讨论】：
标签： python sql sqlalchemy escaping
你不应该尝试实现自己的转义，而应该使用 SQLAlchemy 的内置方法：
sql = 'select * from foo where fieldname = :name'
result = connection.execute(sql, name = myvar)
【讨论】：
INSERT INTO table_name (col1, col2) VALUES ('col1', 'col2) 格式构建聚合插入。 =P 为此，您确实需要在收集值之前真正对其进行转义。
您可以使用 pymysql 中的 escape_string 方法，然后转义 :，这样 SQLAlchemy 就不会尝试为该变量绑定参数，这里是示例
import MySQLdb
query = """ insert into.... values("{}"...) """.format(MySQLdb.escape_string(item).replace(':','\:'))
请注意，如果您使用这种方式，您的代码容易受到 SQL 注入的影响
安装pymysql
pip3 install pymysql
【讨论】：
escape_string 使用反斜杠转义引号，这不适用于 SQLite。
编译和完成其他贡献者的答案。
直接编写 SQL 字符串通常是一个糟糕的解决方案，因为每个数据库系统都支持自己的 SQL 方言，因此 SQL 字符串通常不能跨数据库移植。
为了从用户那里抽象出这个问题，SQLAlchemy 邀请您以更面向对象的方式编写更高级别的 SQL 查询。它被称为 SQL 表达式语言，并在此处记录： https://docs.sqlalchemy.org/en/13/core/tutorial.html
基本上，您可以在 Python 中构建描述您想要执行的操作的表达式，SQLAlchemy 将为您生成相应的 SQL 字符串，并使用您正在使用的数据库的适当方言。
由于您熟悉 SQL，因此您可以在几个小时内学会这种“迷你语言”（我不太愿意在此处加上“s”）。
如果我相信@BrtH，使用这个系统也会为你透明地转义字符串。 转义很难做到正确，因此将其留给成熟的系统总是比尝试自己做要好。
这是一个选择子句的随机示例：
from sqlalchemy import select
...
ham_table = meta.tables['ham']
sel = select([ham_table.c.size, ham_table.c.weight]).where(ham_table.c.taste == 'yummy')
result = meta.bind.execute(sel)
不要被“.c.”所迷惑，它只是一种约定，可以帮助他们为您很好地自动化事情。它们基本上为您生成列描述符，并将它们存储在表对象的 .c 字段下。
【讨论】：
如果必须明确转义字符串，而标准工具不符合要求，您可以要求 SQLAlchemy 使用引擎方言转义。
import sqlalchemy
engine = sqlalchemy.create_engine(...)
sqlalchemy.String('').literal_processor(dialect=engine.dialect)(value="untrusted value")
在我的例子中，我需要根据用户输入动态创建一个数据库（sqlalchemy-utils 具有此功能，但在我的例子中它失败了）。
【讨论】：
扩展@edd 的答案，它在有限的能力范围内起作用。
@edd 提供：
import sqlalchemy
engine = sqlalchemy.create_engine(...)
sqlalchemy.String('').literal_processor(dialect=engine.dialect)(value="untrusted value")
如果您的“不受信任的值”是您要执行的查询，这将导致一个双引号字符串包裹一个单引号字符串，如果不去掉引号就无法直接执行，即"'SELECT ...'"。
您可以使用sqlalchemy.Integer().literal_processor 来做同样的事情，但结果不会有额外的内引号，因为它旨在创建像5 这样的整数，而不是像'5' 这样的字符串。所以你的结果只会被引用一次："SELECT ..."。
我发现这种 Integer 方法有点粗略 - 阅读我的代码的人会知道我为什么要这样做吗？至少对于 psycopg2 来说，有一个更直接、更清晰的方法。
如果您的底层驱动程序是 psycopg2，您可以使用 sqlalchemy 进入驱动程序，获取光标，然后使用 psycopg2 的 cursor.mogrify 绑定和转义您的查询
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
cursor = session.connection().connection.cursor()
processed_query = cursor.mogrify([mogrify args, see docs]).decode("UTF-8")
我知道了如何从这个答案中获取光标：SQLAlchemy, Psycopg2 and Postgresql COPY
并从这个答案中删除：psycopg2 equivalent of mysqldb.escape_string?
我的用例是构建一个查询，然后将其包装在括号中并使用 (SELECT ...) AS temp_some_table 之类的别名，以便将其传递给 PySpark JDBC read。当 SQLAlchemy 构建查询时，它会最小化括号，所以我只能得到SELECT ... AS temp_some_table。我使用上述方法来获得我需要的东西：
cursor = session.connection().connection.cursor()
aliased_query = cursor.mogrify(
f"({query}) AS temp_{model.__tablename__}"
).decode("UTF-8")
【讨论】：
there also existsescape_like
from sqlalchemy_utils import escape_like
User.name.ilike(escape_like('John'))
因此，您可以在将所有字符串格式化为 sql 之前为所有字符串调用 escape_like
【讨论】：