使用SQLAlchemy Core创建用户时如何避免SQL注入？

先回顾Psycopg的安全写法 #

在Psycopg中，我们可以通过SQL、Identifier和Literal的组合避免SQL注入，确保用户名被正确引号包裹，密码被安全处理：

from psycopg.sql import SQL, Identifier, Literal

username = "test_user"
password = "test_password"
query = SQL(
    "CREATE USER {username} WITH ENCRYPTED PASSWORD {password};"
).format(username=Identifier(username), password=Literal(password))

在Connection范围内测试输出：

print(query.as_string(connection))  # CREATE USER "test_user" WITH ENCRYPTED PASSWORD 'test_password';

遇到的问题：SQLAlchemy中quoted_name未生效 #

尝试用quoted_name和绑定参数实现，但用户名没有被正确添加引号：

from sqlalchemy import quoted_name, text

query = text(
    f"CREATE USER {quoted_name(username, True)} WITH ENCRYPTED PASSWORD :password;"
).bindparams(password=password).compile(compile_kwargs={"literal_binds": True})

print(query)  # CREATE USER test_user WITH ENCRYPTED PASSWORD 'test_password';

（修改quoted_name的第二个参数为False或None，结果一致）

正确的实现方式 #

方法1：使用text结合Identifier与绑定参数 #

通过SQLAlchemy的Identifier处理用户名，密码用绑定参数传递，确保两者都被安全处理：

from sqlalchemy import text, create_engine
from sqlalchemy.sql import Identifier

username = "test_user"
password = "test_password"

# 构造查询：用户名用Identifier类型传入，密码用绑定参数
query = text("CREATE USER :username WITH ENCRYPTED PASSWORD :password").bindparams(
    username=Identifier(username),
    password=password
)

# 编译时指定PostgreSQL方言，确保语法正确
compiled_query = query.compile(
    dialect=create_engine("postgresql:///").dialect,
    compile_kwargs={"literal_binds": True}
)
print(compiled_query)  # 输出：CREATE USER "test_user" WITH ENCRYPTED PASSWORD 'test_password'

方法2：使用DDL对象构造语句 #

SQLAlchemy的DDL对象更适合处理数据定义语言语句，配合quoted_name使用：

from sqlalchemy import DDL, create_engine, quoted_name

username = quoted_name("test_user", quote=True)
password = "test_password"

# 构造DDL语句并绑定参数
ddl_stmt = DDL("CREATE USER :username WITH ENCRYPTED PASSWORD :password").bindparams(
    username=username,
    password=password
)

# 编译输出
compiled_stmt = ddl_stmt.compile(
    dialect=create_engine("postgresql:///").dialect,
    compile_kwargs={"literal_binds": True}
)
print(compiled_stmt)  # 输出：CREATE USER "test_user" WITH ENCRYPTED PASSWORD 'test_password'

为什么之前的写法失效？ #

直接用f-string插入quoted_name对象时，Python会直接将其转换为普通字符串，丢失了SQLAlchemy对它的引号处理逻辑。必须通过绑定参数或text的格式化方法传递这类SQL元素，才能让SQLAlchemy在编译时正确处理引号。

内容的提问来源于stack exchange，提问作者Sebastian Thomas