文章目录
- 一、基于 RotatingFileHandler 的自定义处理程序
- 二、基于 TimedRotatingFileHandler 的自定义处理程序
Python logging模块的基本使用、进阶使用详解
Python logging.handlers模块,RotatingFileHandler、TimedRotatingFileHandler 处理器各参数详细介绍
Python logging.config模块,logging.config.fileConfig()、logging.config.dictConfig() 使用介绍
python logging模块Filters过滤器介绍,如何使用自定义的过滤器
这篇文章将介绍如何在logging.config.dictConfig()
日志配置方式下,使用自定义的Handler
处理程序。
一、基于 RotatingFileHandler 的自定义处理程序
默认情况下,使用logging.handlers.RotatingFileHandler()
生成的备份文件,不是以.log
为后缀,例如:原日志文件为rotated.log
,那么生成的备份日志文件为rotated.log.xxx
,xxx
为数字。
接下来将使用自定义处理程序,将生成的备份日志文件的后缀改为rotated.xxx.log
rtfHandler.py
自定义处理程序模块
# -*- coding:utf-8 -*-
import logging
import logging.handlers
import os
import re
def my_rotating_file_handler(filename,mode='a',maxBytes=0,backupCount=0,encoding=None,delay=False):
rh = logging.handlers.RotatingFileHandler(
filename,
mode=mode,
maxBytes=maxBytes,
backupCount=backupCount,
encoding=encoding,
delay=delay
)
global bckCount # 声明全局变量,供 remove_old_log()、rotator() 方法使用
bckCount = backupCount
# 调用新实现的备份日志文件命名规则
rh.rotator = rotator
rh.namer = namer
return rh
def namer(name):
# 在备份文件后增加 .log 后缀
name = name + '.log'
return name
def rotator(source, dest):
global dest_file # 声明全局变量,供 remove_old_log() 方法使用
dest_file = dest
sp = dest.rsplit(".",3)
sp.pop(1) # 删除备份文件中间的那个".log"
# 因为改了备份日志文件名称规则,所以原有的备份规则失效了,下面是重写的备份规则
for dst in range(bckCount,0,-1):
sp[1] = str(dst)
dest = ".".join(sp)
if os.path.exists(dest):
os.remove(dest)
sp[1] = str(dst-1)
src = ".".join(sp)
if os.path.exists(src):
os.rename(src,dest)
sp[1] = str(1)
dest = ".".join(sp)
os.rename(source,dest)
remove_old_log()
# 因为改变了命名规则,所以重新实现了一个删除旧日志文件的方法
def remove_old_log():
logdir = os.path.dirname(dest_file)
fname = os.path.basename(dest_file)
flist = os.listdir(logdir)
fsplit = fname.rsplit(".",3)
pattern = re.compile(r".".join([fsplit[0],"[0-9]+",fsplit[3]]))
for f in flist:
if re.match(pattern,f) and int(f.rsplit(".",2)[1]) > bckCount:
fpath = os.path.join(logdir,f)
os.remove(fpath)
rtfLog.yaml
日志配置文件,handlers.rtfhandler
下配置了一个自定义的处理程序
version: 1
disable_existing_loggers: false
formatters:
simple:
format: '[%(asctime)s - %(name)s - %(levelname)-8s] %(message)s'
handlers:
console:
class: logging.StreamHandler
formatter: simple
level: DEBUG
stream: ext://sys.stdout
rtfhandler:
# 通过()键,引用一个自定义的处理程序
# ext:// 用来告诉dicConfig(),它后面的config.rtfHandler.my_rotating_file_handler 是配置以外的对象
(): ext://config.rtfHandler.my_rotating_file_handler
level: DEBUG
formatter: simple
# 下面的标量都是提供给自定义的处理程序方法的参数
filename: rotated.log
mode: a
maxBytes: 40
backupCount: 2
encoding: utf8
delay: False
loggers:
mylogger:
level: DEBUG
handlers: [console,rtfhandler]
propagate: no
rtfLogger.py
封装日志模块,使用dictConfig()
读取日志配置文件以完成日志配置
# -*- coding:utf-8 -*-
import logging.config
import yaml
import os
class Logger():
def __init__(self,logger_name=''):
self.logger_name = logger_name
# 获取当前文件所在目录
cur_dir = os.path.dirname(__file__)
# 拼接对应的yaml配置文件路径
log_conf = os.path.join(cur_dir, 'logconf', 'rtflog.yaml')
with open(log_conf, 'r', encoding='utf8') as f:
config = yaml.safe_load(f.read())
# 加载配置文件
logging.config.dictConfig(config)
def logger(self):
logger = logging.getLogger(self.logger_name)
return logger
main.py
主程序,测试日志功能
# -*- coding:utf-8 -*-
from config.rtfLogger import Logger
import time
mlogger = Logger('mylogger').logger()
for i in range(10):
mlogger.info(f'Message no. {i + 1}')
time.sleep(1)
运行main.py
程序,输出:
二、基于 TimedRotatingFileHandler 的自定义处理程序
默认情况下,使用logging.handlers.TimedRotatingFileHandler ()
生成的备份文件,不是以.log
为后缀,例如:原日志文件为timed_rotated.log
,那么生成的备份日志文件为timed_rotated.log.xxx
,xxx
为格式化的日期时间,如:2023-08-21_23-56-50
。
接下来将使用自定义处理程序,将生成的备份日志文件的后缀改为rotated.xxx.log
trfHandler.py
自定义处理程序模块
# -*- coding:utf-8 -*-
import logging
import logging.handlers
import os
import re
def my_timed_rotating_file_handler(filename,when='h',interval=1,backupCount=0,encoding=None,delay=False,utc=False,atTime=None):
global trfh
trfh = logging.handlers.TimedRotatingFileHandler(
filename=filename,
when=when,
interval=interval,
backupCount=backupCount,
encoding=encoding,
delay=delay,
utc=utc,
atTime=atTime
)
trfh.rotator = rotator
trfh.namer = namer
return trfh
def namer(name):
name = name + '.log'
return name
# 重新定义了备份文件命名规则
def rotator(source, dest):
fsp = dest.rsplit(".",3)
fsp.pop(1)
dest = ".".join(fsp)
os.rename(source,dest)
remove_old_log(trfh)
# 重新定义了旧备份文件删除规则
def remove_old_log(trfh):
logdir = os.path.dirname(trfh.baseFilename)
fname = os.path.basename(trfh.baseFilename)
pattern = re.compile(fname.rsplit(".",1)[0] + "\." + trfh.extMatch.pattern[1:],re.ASCII)
flist = os.listdir(logdir)
loglist = []
for f in flist:
if re.match(pattern,f):
loglist.append(f)
loglist.sort(reverse=True)
backupCount = trfh.backupCount
log_to_remove = loglist[backupCount:] if len(loglist) > backupCount else []
for ltr in log_to_remove:
os.remove(os.path.join(logdir,ltr))
trfLog.yaml
日志配置文件,handlers.trfhandler
下配置了一个自定义的处理程序
version: 1
disable_existing_loggers: false
formatters:
simple:
format: '[%(asctime)s - %(name)s - %(levelname)-8s] %(message)s'
handlers:
console:
class: logging.StreamHandler
formatter: simple
level: DEBUG
stream: ext://sys.stdout
trfhandler:
# 通过()键,引用一个自定义的处理程序
# ext:// 用来告诉dicConfig(),它后面的config.rtfHandler.my_rotating_file_handler 是配置以外的对象
(): ext://config.trfHandler.my_timed_rotating_file_handler
level: DEBUG
formatter: simple
# 下面的标量都是提供给自定义的处理程序方法的参数
filename: timed_rotated.log
when: S
interval: 2
backupCount: 2
encoding: utf8
delay: False
utc: False
atTime: None
loggers:
mylogger:
level: DEBUG
handlers: [console,trfhandler]
propagate: no
trfLogger.py
封装日志模块,使用dictConfig()
读取日志配置文件以完成日志配置
# -*- coding:utf-8 -*-
import logging.config
import yaml
import os
class Logger():
def __init__(self,logger_name=''):
self.logger_name = logger_name
cur_dir = os.path.dirname(__file__)
log_conf = os.path.join(cur_dir, 'logconf', 'trflog.yaml')
with open(log_conf, 'r', encoding="utf8") as f:
config = yaml.safe_load(f.read())
logging.config.dictConfig(config)
def logger(self):
logger = logging.getLogger(self.logger_name)
return logger
main.py
主程序,测试日志功能
# -*- coding:utf-8 -*-
import logging
from config.trfLogger import Logger
import time
mlogger = Logger('mylogger').logger()
for i in range(10):
mlogger.info(f'Message no. {i + 1}')
time.sleep(1)
运行main.py
程序,输出:
参考资料:
using-a-rotator-and-namer-to-customize-log-rotation-processing
customizing-handlers-with-dictconfig
logging-config-dict-externalobj