asyncpg 禁用 f-string 拼接 sql 防注入，因不支持标识符参数化，表名列名须白名单校验或安全转义；where 动态条件需分离 sql 片段与参数列表；执行前须检查连接状态、事务上下文并捕获 asyncpg.postgreserror。
asyncpg 为什么不能直接用 f-string 拼接 SQL
因为 asyncpg 默认禁用字符串拼接式查询，一旦用 f"SELECT * FROM {table_name}"，就可能触发 SQL 注入——哪怕你“确认输入可控”，asyncpg 也不买账。它底层不走服务端预编译（像 PostgreSQL 的 PREPARE），而是靠客户端参数化 + 服务端协议级绑定，所以所有动态部分必须显式声明为参数或通过白名单校验。
表名/列名等非参数化部分必须白名单校验
SQL 中的标识符（如表名、字段名、ORDER BY 字段）无法用 $1 占位符代替，asyncpg 不支持标识符参数化。唯一安全做法是提前定义允许值集合，并严格比对：
- 用
enum.Enum或frozenset存储合法表名，例如ALLOWED_TABLES = frozenset({"users", "orders", "products"}) - 校验时用
if table_name not in ALLOWED_TABLES: raise ValueError("Invalid table") - 拼接前转义：若必须支持少量动态标识符，可用
asyncpg.utils._quote_ident(table_name)（注意这是私有函数，生产环境建议自己实现简单引号包裹：f'"{table_name.replace('"', '""')}"'，且仍需配合白名单）
WHERE 条件动态组合要用字典 + 构建参数列表
不要手写 "WHERE " + " AND ".join(conds)。正确做法是维护 SQL 片段与参数分离的结构：
conditions = []
params = []
param_idx = 1
<p>if user_id:
conditions.append(f"id = ${param_idx}")
params.append(user_id)
param_idx += 1
if status:
conditions.append(f"status = ${param_idx}")
params.append(status)
param_idx += 1</p><p>where_clause = " AND ".join(conditions) if conditions else "TRUE"
sql = f"SELECT <em> FROM users WHERE {where_clause}"
await conn.fetch(sql, </em>params)</p>
这样既保持参数化，又避免空条件导致语法错误。注意：asyncpg 要求参数顺序与 $1, $2 严格对应，不能跳号或重复使用同一编号。
执行前务必检查连接状态和事务上下文
asyncpg 的 Connection 对象不是线程安全的，且在异常中断后可能处于不可用状态。动态 SQL 尤其容易因拼接逻辑出错导致语句非法，进而让连接卡在坏状态：
- 永远用
try/except asyncpg.PostgresError包裹执行逻辑，而不是裸奔except Exception - 在事务中执行动态 SQL 时，确保上层已显式调用
conn.transaction()并正确__aenter__/__aexit__，否则conn.execute()可能静默失败或污染事务隔离 - 对高风险动态语句（如 DDL），加
timeout参数：await conn.execute(sql, *params, timeout=5.0)
最易被忽略的是：asyncpg 不会自动重试或恢复损坏的连接。一次 SyntaxError 后，该连接大概率不能再复用，得手动 close() 并重建。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
