在SQLAlchemy中手动构建SQL查询时如何正确转义字符串？
我在用SQLAlchemy这个工具在Python里连接不同的数据库，不过我没有用它的ORM功能，因为有几个原因让我不能这样做。
主要是我需要构建一个复杂的SQL查询，使用一些像下面这样的东西：
sql += "AND fieldname = '%s'" % myvar
在我的情况下，不用担心SQL注入的问题，因为数据总是来自可信的来源。但是即使来源是可信的，里面也可能有一些字符会导致查询出错，比如 '、% 或 _。
所以我主要需要对这些字符进行转义，我在想有没有现成的转义函数可以重复使用。
6 个回答
你可以使用pymysql里的escape_string方法，然后对:进行转义，这样SQLAlchemy就不会试图为那个变量绑定参数了，下面是一个示例
import MySQLdb
query = """ insert into.... values("{}"...) """.format(MySQLdb.escape_string(item).replace(':','\:'))
请注意，如果你这样做，你的代码可能会受到SQL注入攻击的威胁
要安装pymysql
pip3 install pymysql
这里还有一个 escape_like
from sqlalchemy_utils import escape_like
User.name.ilike(escape_like('John'))
所以在把你的字符串格式化成SQL之前，可以先调用一下escape_like。
如果你不想添加额外的第三方库，你可以这样做：
def escape_sql_search_text(text: str) -> str:
return text.replace("%", "\\%").replace("\\", "\\\\").replace("_", "\\_")
User.name.ilike(escape_sql_search_text('John'))
如果你想在SQL中使用f-string（不推荐这样做），那么你还需要用 .replace("'", "''") 来处理一下。
def _escape_sql_search_text_for_fstring(text: str) -> str:
return text.replace("%", "\\%").replace("\\", "\\\\").replace("_", "\\_").replace("'", "''")
ilike_text = _escape_sql_search_text_for_fstring(text)
sql = f"name::varchar ilike '{ilike_text}'"
为了补充一下@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 ..."。
我觉得这个整数的方法有点不太靠谱——看我代码的人会知道我为什么这么做吗？至少对于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
而mogrify是从这个回答中学到的： 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")
有时候，我们需要特别处理一个字符串，让它变得安全，但标准的方法不太合适。这时，你可以让 SQLAlchemy 使用数据库引擎的方言来处理这个字符串。
import sqlalchemy
engine = sqlalchemy.create_engine(...)
sqlalchemy.String('').literal_processor(dialect=engine.dialect)(value="untrusted value")
在我的例子中，我需要根据用户输入动态创建一个数据库（虽然 sqlalchemy-utils 有这个功能，但在我的情况下它没能成功）。
你不应该自己去实现转义功能，而是应该使用SQLAlchemy自带的方法：
sql = 'select * from foo where fieldname = :name'
result = connection.execute(sql, name = myvar)