Psycopg3异步模式下，如何让字符串类型的SQL查询兼容cursor.execute方法？

我完全懂你的困扰——好不容易搭好的查询生成器不想动，结果切换到psycopg3异步版本，就因为SQL字符串的类型问题卡壳了。这个报错本质是psycopg3为了SQL注入防护，对传入execute的查询类型做了更严格的限制：它不再直接接受普通的str，而是要求使用它提供的SQL对象、字面量安全字符串（LiteralString）或者其他兼容的Query类型。

不过别担心，我们不需要改查询生成器，只要加一步简单的包装就能解决问题。

解决方案：用psycopg.sql.SQL包装字符串查询 #

psycopg提供了SQL类来封装动态生成的SQL字符串，它能被execute方法识别为合法的Query类型。我们只需要做两个小修改：

1. 导入psycopg.sql模块里的SQL类
2. 在调用execute时，把查询生成器返回的字符串用SQL()包裹起来

修改后的完整代码 #

import asyncio
import psycopg
from psycopg.sql import SQL  # 新增导入SQL类

class MyQueryBuilderObj:
    def getFinalQuery(self) -> str:
        return "SELECT * FROM test_table"

async def fetch_data_from_db():
    qb_obj = MyQueryBuilderObj()
    async with await psycopg.AsyncConnection.connect("dbname=test user=postgres password=yourpassword host=localhost") as aconn:
        async with aconn.cursor() as acur:
            # 用SQL()包装动态生成的查询字符串，兼容psycopg3的类型要求
            await acur.execute(SQL(qb_obj.getFinalQuery()))
            result = await acur.fetchone()
            return result

async def main():
    try:
        result = await fetch_data_from_db()
        print("Fetched data:", result)
    except Exception as e:
        print("An error occurred:", e)

if __name__ == "__main__":
    asyncio.run(main())

针对你提到的带参数的场景（查询含%s+参数列表） #

你提到实际中查询生成器会输出带%s占位符的SQL和对应的参数列表，这种情况同样可以用SQL处理，直接把参数列表传给execute的第二个参数即可：

# 假设查询生成器返回查询字符串和参数列表
query_str, params = qb_obj.getFinalQueryWithParams()
await acur.execute(SQL(query_str), params)

关于类型错误的补充说明 #

你遇到的报错信息：

Argument of type "str" cannot be assigned to parameter "query" of type "Query" in function "execute"
Type "str" is not assignable to type "Query"
"str" is not assignable to "LiteralString"
"str" is not assignable to "bytes"
"str" is not assignable to "SQL"
"str" is not assignable to "Composed"

这是psycopg3的安全设计：它希望开发者避免直接传入可能存在注入风险的动态拼接字符串，而通过SQL对象，你可以明确告诉psycopg：这个动态生成的查询是安全的（前提是你的查询生成器没有直接拼接用户输入的未转义内容）。

备注：内容来源于stack exchange，提问作者Fogux