如何获取asyncpg执行的填充参数后的完整SQL查询语句？

核心结论 #

asyncpg 本身没有提供你期望的connection.expected_query这类方法，原因是 asyncpg 采用 PostgreSQL 服务端预处理语句协议，查询模板和参数是分开传输到数据库服务端的，客户端不会提前拼接完整SQL，既可以避免SQL注入风险，也能提升查询性能。你可以通过以下几种方案满足调试需求：

方案1：开启PostgreSQL服务端日志（结果最准确） #

修改PostgreSQL配置文件postgresql.conf，调整以下参数：

log_statement = 'all'
log_min_messages = debug1

重启PostgreSQL服务后，所有实际执行的完整SQL语句都会被记录到日志文件中，参数是服务端实际替换后的结果，完全准确无转义误差。

方案2：自定义调试用SQL拼接函数（无需实际执行即可查看） #

你可以自己实现一个简单的参数替换函数，适配PostgreSQL的$N占位符规则，调试时传入和execute完全一致的入参即可得到拼接后的SQL，示例实现如下：

def build_debug_pg_query(query_template: str, *args) -> str:
    processed_params = []
    for arg in args:
        if isinstance(arg, str):
            # 转义字符串中的单引号
            escaped_val = arg.replace("'", "''")
            processed_params.append(f"'{escaped_val}'")
        elif arg is None:
            processed_params.append("NULL")
        elif isinstance(arg, (int, float, bool)):
            processed_params.append(str(arg))
        # 其他类型比如日期、二进制等可以根据需求自行扩展处理逻辑
        else:
            processed_params.append(f"'{str(arg)}'")
    # 按顺序替换占位符
    result_query = query_template
    for idx, param in enumerate(processed_params, start=1):
        result_query = result_query.replace(f"${idx}", param)
    return result_query

使用方式和你期望的expected_query完全一致：

query = QUERY_1
param1 = query_params_dict["field_1"]
param2 = int(query_params_dict["field_id"])
# 调试时直接打印查看，不需要执行查询
print(build_debug_pg_query(query, param1, param2))
# 实际执行还是用原生参数绑定，避免注入风险
await connection.execute(query, param1, param2)

注意：这个函数仅用于调试排查问题，不要在生产环境用它拼接SQL后直接执行，会存在SQL注入风险。

方案3：开启asyncpg的DEBUG日志 #

无需修改数据库配置，只需调整日志级别即可看到分开记录的查询模板和对应参数：

import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("asyncpg").setLevel(logging.DEBUG)

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