我已经连接到PostgreSQL数据库:
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
# ...
当我执行查询时，它以字节的形式记录到LOG文件中，而不是随意的字符串，因此不专门显示西里尔字符。下面是一个例子:
2022-12-15 03:47:59,914 DEBUG [extras.py:427] b"INSERT INTO test VALUES (1, 'xd0x9axd0xb8xd1x80xd0xb8xd0xbbxd0xbbxd0xb8xd1x86xd0xb0')"
对应于:INSERT INTO test VALUES (1, 'Кириллица')(只是一些测试数据)。
那么有什么方法可以解码字节吗?记录与执行查询无关的其他事件(DEBUG)。
提前感谢。
期待这样的事情:
2022-12-15 03:47:59,914 DEBUG [extras.py:427] INSERT INTO test VALUES (1, 'Кириллица')
附言我决定不使用LoggingConnection，并创建了执行查询的方法，并以我想要的方式记录它们:
def execute(self, query: str) -> tuple | None:
"""
"""
from contextlib import closing
with closing(self.connection) as connection:
with connection.cursor() as cursor:
try:
cursor.execute(query)
LOGGER.debug(query)
except:
# catch my exceptions here
我不知道这个解决方案是否有效，但它对我来说非常有效。
您可以向记录器添加一个过滤器来解码字节:
class DecodingFilter:
# Provide the correct encoding if your connection is not using UTF-8.
def __init__(self, encoding='utf-8'):
self.encoding = encoding
def filter(self, record):
record.msg = record.msg.decode(self.encoding)
return record
logger.addFilter(DecodingFilter())
logger = logging.getLogger('test_logger')
logger.addFilter(DecodingFilter())
logger.setLevel(logging.DEBUG)
logging.basicConfig()
with psycopg2.connect(dbname='test', connection_factory=extras.LoggingConnection) as conn:
conn.initialize(logger)
cur = conn.cursor()
cur.execute('SELECT 1')
生成如下输出:
DEBUG:test_logger:SELECT 1