Psycopg2中DEBUG级查询日志以字节记录，如何解码显示正常字符？

使用psycopg2的LoggingConnection连接PostgreSQL时，执行查询的DEBUG日志会以字节形式（如b"INSERT INTO test VALUES (1, '\xd0\x9a\xd0\xb8\xd1\x80\xd0\xb8\xd0\xbb\xd0\xbb\xd0\xb8\xd1\x86\xd0\xb0')"）写入日志文件，导致西里尔字符无法正常显示为Кириллица。非查询相关的DEBUG日志显示正常。

当前连接代码：

LOGGER = Logger().LOGGER
# ...
self.connection = psycopg2.connect(connection_factory=LoggingConnection,
                                   dbname=...,
                                   user=...,
                                   password=...,
                                   host=...,
                                   options="-c search_path=...")
self.connection.initialize(LOGGER)
self.cursor = self.connection.cursor()

自定义Logger类：

# some data

_MODE = "w"
_ENCODING = "utf-8"
_FORMATTER = "%(asctime)s %(levelname)s [%(filename)s:%(lineno)s] %(message)s"

class Logger:
    @property
    def LOGGER(self):
        logger = logging.getLogger()
        logger.setLevel(logging.DEBUG)

        fileHandler = logging.FileHandler(filename=_FILE,
                                          mode=_MODE,
                                          encoding=_ENCODING)
        fileHandler.setFormatter(logging.Formatter(_FORMATTER))
        logger.addHandler(fileHandler)

        return logger
# ...

期望日志格式：

2022-12-15 03:47:59,914 DEBUG [extras.py:427] INSERT INTO test VALUES (1, 'Кириллица')

方法一：重写LoggingConnection的日志输出方法 #

psycopg2的LoggingConnection默认会把SQL语句以字节形式传递给日志器，我们可以自定义继承类，重写其日志方法完成解码：

import psycopg2
from psycopg2.extras import LoggingConnection

class DecodedLoggingConnection(LoggingConnection):
    def _log_message(self, msg, cur):
        # 将字节消息解码为UTF-8字符串
        if isinstance(msg, bytes):
            msg = msg.decode('utf-8')
        # 调用父类原日志逻辑
        super()._log_message(msg, cur)

修改连接代码，使用自定义的连接工厂：

self.connection = psycopg2.connect(connection_factory=DecodedLoggingConnection,
                                   dbname=...,
                                   user=...,
                                   password=...,
                                   host=...,
                                   options="-c search_path=...")
self.connection.initialize(LOGGER)
self.cursor = self.connection.cursor()

方法二：给日志器添加字节解码过滤器 #

如果不想修改连接类，可以给日志器添加全局过滤器，自动处理所有字节类型的日志消息：

class ByteDecodeFilter(logging.Filter):
    def filter(self, record):
        # 解码日志消息主体
        if isinstance(record.msg, bytes):
            record.msg = record.msg.decode('utf-8')
        # 解码日志参数（若存在）
        if record.args:
            decoded_args = []
            for arg in record.args:
                decoded_args.append(arg.decode('utf-8') if isinstance(arg, bytes) else arg)
            record.args = tuple(decoded_args)
        return True

在自定义Logger类中添加该过滤器：

class Logger:
    @property
    def LOGGER(self):
        logger = logging.getLogger()
        logger.setLevel(logging.DEBUG)
        # 添加字节解码过滤器
        logger.addFilter(ByteDecodeFilter())

        fileHandler = logging.FileHandler(filename=_FILE,
                                          mode=_MODE,
                                          encoding=_ENCODING)
        fileHandler.setFormatter(logging.Formatter(_FORMATTER))
        logger.addHandler(fileHandler)

        return logger

说明 #

两种方法均可将日志中的字节消息解码为UTF-8字符串，实现西里尔字符的正常显示。方法一直接针对psycopg2的日志逻辑，更精准；方法二更通用，可覆盖其他场景下的字节日志处理。

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