如何在执行前获取asyncpg生成的查询语句的原始SQL?
asyncpg 如何将带$n参数的查询模板转换为最终SQL文本?
asyncpg 并没有提供公开的官方函数来直接将带$n占位符的查询模板与参数拼接成最终的SQL文本。这是因为asyncpg的设计逻辑是将查询模板和参数分开发送给PostgreSQL数据库,由数据库完成参数绑定,以此避免客户端拼接SQL带来的SQL注入风险。
如果只是用于日志记录(注意:绝对不能用生成的SQL去执行,否则会引入注入风险),可以通过以下两种方式实现:
1. 自定义简单替换函数
针对常见的数据类型,可以自己实现一个替换逻辑:
def format_query(query_tpl, *values): formatted_values = [] for val in values: # 根据数据类型做简单格式化,比如字符串加单引号,数字直接转字符串 if isinstance(val, str): # 转义单引号避免SQL语法错误 formatted_values.append(f"'{val.replace('''', '''''')}'") elif isinstance(val, (int, float)): formatted_values.append(str(val)) elif val is None: formatted_values.append("NULL") # 可扩展更多数据类型的处理逻辑 else: formatted_values.append(repr(val)) # 替换$1、$2等占位符 for idx, val_str in enumerate(formatted_values, 1): query_tpl = query_tpl.replace(f"${idx}", val_str) return query_tpl # 测试示例 query_tpl = "SELECT $1, $2, $3" values = (42, 'hello world', None) print(format_query(query_tpl, *values)) # 输出: SELECT 42, 'hello world', NULL
2. 使用asyncpg内部的参数编码工具
asyncpg内部有用于编码参数的模块,可以借助它来更准确地格式化参数,但要注意这是内部API,未来版本可能会发生变化:
import asyncpg.protocol.codecs as codecs def format_query_with_asyncpg_codecs(query_tpl, *values): formatted_values = [] for val in values: if val is None: formatted_values.append("NULL") else: # 自动推断数据类型并编码为文本格式 dtype = codecs.infer_type(val) encoder = codecs.get_codec(dtype, format='text') encoded = encoder(val, None) formatted_values.append(encoded.decode('utf-8')) # 替换占位符 for idx, val_str in enumerate(formatted_values, 1): query_tpl = query_tpl.replace(f"${idx}", val_str) return query_tpl # 测试示例 query_tpl = "SELECT $1, $2" values = (42, 'test with ''quote''') print(format_query_with_asyncpg_codecs(query_tpl, *values)) # 输出: SELECT 42, 'test with ''quote'''
重要提醒:无论用哪种方式生成的SQL,都只能用于日志记录,不能用于执行查询,否则会破坏asyncpg的参数化查询安全机制,引发SQL注入漏洞。
内容的提问来源于stack exchange,提问作者Fomalhaut
相关产品推荐
相关产品推荐

