psycopg3中Cursor无mogrify属性问题及替代方案咨询

问题说明 #

你使用的psycopg3（3.1.18版本）的Cursor对象确实没有mogrify方法——这个方法在psycopg2中属于Cursor，但psycopg3将其迁移到了Connection对象上，同时也提供了更灵活的SQL字符串构建方式。

替代方案 #

方案一：直接使用Connection对象的mogrify方法

这是最接近psycopg2中mogrify用法的方式，无需创建Cursor，直接通过连接对象生成填充参数后的SQL：

import logging
import psycopg
import pandas as pd

# 修改后的代码
with psycopg.connect(self.connect_str, autocommit=True) as conn:
    if self.log.level == logging.DEBUG:
        # 调用conn的mogrify方法
        sql_mogr = conn.mogrify(sql, params)
        # 转成字符串输出，避免日志显示bytes类型
        self.log.debug(sql_mogr.decode('utf-8'))
    else:
        self.log.info(f'sql: {sql}, params:{params}')

    df = pd.read_sql(sql, con=conn, params=params)

方案二：使用psycopg.sql模块构建SQL字符串

如果需要处理动态表名、字段名等更复杂的场景，推荐使用psycopg.sql模块，它能确保SQL拼接的安全性，避免注入风险：

import logging
import psycopg
from psycopg import sql
import pandas as pd

# 修改后的代码
with psycopg.connect(self.connect_str, autocommit=True) as conn:
    if self.log.level == logging.DEBUG:
        # 根据参数类型选择格式化方式
        if isinstance(params, dict):
            # 命名参数场景
            query = sql.SQL(sql).format(**{k: sql.Literal(v) for k, v in params.items()})
        else:
            # 位置参数场景
            query = sql.SQL(sql).format(*map(sql.Literal, params))
        # 生成可直接执行的SQL字符串
        sql_mogr = query.as_string(conn)
        self.log.debug(sql_mogr)
    else:
        self.log.info(f'sql: {sql}, params:{params}')

    df = pd.read_sql(sql, con=conn, params=params)

注意事项 #

• conn.mogrify返回的是bytes类型，必须用.decode('utf-8')转换为字符串后再输出日志，否则会显示类似b'SELECT ...'的内容。
• 不要手动拼接SQL字符串，始终使用参数化查询或psycopg.sql模块处理，避免SQL注入风险。

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