失败时的 Python 日志记录转储



我正在使用Python编写一个从消息队列中读取的服务。对于每条消息,它都会运行一个进程,完成后将从队列中获取另一条消息。我正在使用日志记录包来记录信息、警告和错误。如果进程因任何原因失败,进程将捕获错误并发送一封包含回溯的电子邮件。我希望电子邮件还包含应用程序开始处理消息时的日志。需要明确的是 - 我仍然希望应用程序立即登录。但是,如果应用程序失败并成功捕获异常,我希望它也在电子邮件中发送日志。

我的第一个想法是使用一个只保存日志字符串的列表,如下所示:

while True:
# Grab the message here
log_list = []
try:
log_msg = "Some log message here"
log_list.append(log_msg)
...
except:
# send email here using log_list

这将解决我的问题,除了我想看到python日志记录包添加到日志消息中的其他信息,例如时间戳。我意识到我可以手动将其放入并将其添加到我的log_list对象中。我想做的是:

import logging
logger = logging.getLogger()
while True:
logger.reset()
# Grab the message here
try:
logger.info("Some log message here")
...
except:
# send email here using something like logger.dump()

所以......我的问题 - 有没有另一种方法可以避免使用列表来保存我的日志并保存其他日志记录信息,例如时间戳、日志记录级别等?

只尝试自定义日志类? 像这样:

import time
class Log:
msgs = []
def __init__(self):
pass
def addLog(self,msg):
self.msgs.append("LOG AT "+str(time.time())+":n"+msg)
def dumpLog(self):
s = ""
for each in self.msgs:
s+= each+"n------------------------n"
return s

我不熟悉日志记录模块,对于大多数用例来说,它看起来很复杂。

我最终使用了riscnotcisc的建议,但我扩展了日志记录。记录器类

from logging import Logger
import logging
import datetime
class CustomLogger(Logger):
def __init__(self, process_name):
self.process_name = process_name
super(CustomLogger, self).__init__(process_name)
self.log_list = []
ch = logging.StreamHandler()
ch.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
self.handlers.append(ch)
def info(self, message, *args, **kwargs):
self.log_list.append('{} - {} - INFO - {}'.format(datetime.datetime.now(), self.process_name, message))
super(CustomLogger, self).info(message, *args, **kwargs)
def debug(self, message, *args, **kwargs):
self.log_list.append('{} - {} - DEBUG - {}'.format(datetime.datetime.now(), self.process_name, message))
super(CustomLogger, self).debug(message, *args, **kwargs)
def warning(self, message, *args, **kwargs):
self.log_list.append('{} - {} - WARNING - {}'.format(datetime.datetime.now(), self.process_name, message))
super(CustomLogger, self).warning(message, *args, **kwargs)
def dump(self):
return 'n'.join(self.log_list)
def reset(self):
self.log_list = []

最新更新