建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Python如何僅用5000行代碼，實現(xiàn)強大的logging模塊？
Python 的 logging 模塊實現(xiàn)了靈活的日志系統(tǒng)。整個模塊僅僅 3 個類，不到 5000 行代碼的樣子，學(xué)習(xí)它可以加深對程序日志的了解，本文分下面幾個部分:
-
logging 簡介
-
logging API 設(shè)計
-
記錄器對象 Logger
-
日志記錄對象 LogRecord
-
處理器對象 Hander
-
格式器對象 Formatter
-
滾動日志文件處理器
-
小結(jié)
-
小技巧
logging 簡介
本次代碼使用的是 python 3.8.5 的版本，官方中文文檔 3.8.8 。參考鏈接中官方中文文檔非常詳細，建議先看一遍了解日志使用。
| 類 | 功能 |
|---|---|
| logging-module | logging的API |
| Logger | 日志記錄器對象類，可以創(chuàng)建一個對象用來記錄日志 |
| LogRecord | 日志記錄對象，每條日志記錄都封裝成一個日志記錄對象 |
| Hander | 處理器對象，負責日志輸出到流/文件的控制 |
| Formatter | 格式器，負責日志記錄的格式化 |
| RotatingFileHandler | 按大小滾動的日志文件記錄器 |
| TimedRotatingFileHandler | 按時間滾動的日志文件處理器 |
我們主要研究日志如何輸出到標準窗口這一主線；日志的配置，日志的線程安全及各種特別的Handler等支線可以先忽略。
logging API 設(shè)計
先看看日志使用：
- import logging
- logging.basicConfig(level=logging.INFO, format='%(levelname)-8s %(name)-10s %(asctime)s %(message)s')
- lang = {"name": "python", "age":20}
- logging.info('This is a info message %s', lang)
- logging.debug('This is a debug message')
- logging.warning('This is a warning message')
- logger = logging.getLogger(__name__)
- logger.warning('This is a warning')
輸出內(nèi)容如下:
- INFO root 2021-03-04 00:03:53,473 This is a info message {'name': 'python', 'age': 20}
- WARNING root 2021-03-04 00:03:53,473 This is a warning message
- WARNING __main__ 2021-03-04 00:03:53,473 This is a warning
可以看到 logging 的使用非常方便，模塊直接提供了一組API。
- root = RootLogger(WARNING) # 默認提供的logger
- Logger.root = root
- Logger.manager = Manager(Logger.root)
- def debug(msg, *args, **kwargs): # info，warning等api類似
- if len(root.handlers) == 0:
- basicConfig() # 默認配置
- root.debug(msg, *args, **kwargs)
- def getLogger(name=None):
- if name:
- return Logger.manager.getLogger(name) # 創(chuàng)建特定的logger
- else:
- return root # 返回默認的logger
這種API的提供方式，我們在 requests 中也有看到。api中很重要的設(shè)置config的方式:
- def basicConfig(**kwargs):
- ...
- if handlers is None:
- filename = kwargs.pop("filename", None)
- mode = kwargs.pop("filemode", 'a')
- if filename:
- h = FileHandler(filename, mode)
- else:
- stream = kwargs.pop("stream", None)
- h = StreamHandler(stream) # 默認的handler
- handlers = [h]
- dfs = kwargs.pop("datefmt", None)
- style = kwargs.pop("style", '%')
- fs = kwargs.pop("format", _STYLES[style][1])
- fmt = Formatter(fs, dfs, style) # 生成formatter
- for h in handlers:
- if h.formatter is None:
- h.setFormatter(fmt)
- root.addHandler(h) # 設(shè)置root的handler
- level = kwargs.pop("level", None)
- if level is not None:
- root.setLevel(level) # 設(shè)置日志級別
可以看到，日志的配置主要包括下面幾項:
-
level 日志級別
-
format 信息格式化模版
-
filename 輸出到文件
- datefmt
%Y-%m-%d %H:%M:%S,uuu時間的格式模版 - style [
%,{,$] 格式樣板
演示代碼輸出中，可以看到debug日志沒有顯示，是因為 debug < info :
- CRITICAL = 50
- FATAL = CRITICAL
- ERROR = 40
- WARNING = 30
- WARN = WARNING
- INFO = 20
- DEBUG = 10
- NOTSET = 0
記錄器對象 Logger
查看Logger之前，先看logger對象的管理類Manager
- _loggerClass = Logger
- class Manager(object):
- def __init__(self, rootnode):
- self.root = rootnode
- self.disable = 0
- self.loggerDict = {} # 所有日志記錄對象的字典
- ...
- def getLogger(self, name):
- rv = None
- if name in self.loggerDict:
- rv = self.loggerDict[name] # 獲取已經(jīng)創(chuàng)建過的同名logger
- ...
- else:
- rv = (self.loggerClass or _loggerClass)(name) # 創(chuàng)建新的logger
- rv.manager = self
- self.loggerDict[name] = rv
- ...
- return rv
日志過濾器
- class Filterer(object):
- def __init__(self):
- self.filters = []
- def addFilter(self, filter):
- self.filters.append(filter)
- def removeFilter(self, filter):
- self.filters.remove(filter)
- def filter(self, record):
- rv = True
- for f in self.filters: # 過濾日志
- if hasattr(f, 'filter'):
- result = f.filter(record)
- else:
- result = f(record) # assume callable - will raise if not
- if not result:
- rv = False
- break
- return r
核心的 Logger 實際上只是一個控制中心：
- class Logger(Filterer): # logger可以過濾日志
- def __init__(self, name, level=NOTSET):
- Filterer.__init__(self)
- self.name = name
- self.level = _checkLevel(level)
- self.parent = None # 日志可以有層級
- self.propagate = True
- self.handlers = [] # 可以輸出到多個handler
- self.disabled = False # 可以關(guān)閉
- self._cache = {}
- def debug(self, msg, *args, **kwargs): # 輸出debug日志
- if self.isEnabledFor(DEBUG):
- self._log(DEBUG, msg, args, **kwargs)
logger可以判斷日志級別:
- def isEnabledFor(self, level):
- if self.disabled:
- return False
- try:
- return self._cache[level]
- except KeyError:
- try:
- if self.manager.disable >= level:
- is_enabled = self._cache[level] = False
- else:
- is_enabled = self._cache[level] = (
- level >= self.getEffectiveLevel()
- )
- return is_enabled
- def getEffectiveLevel(self):
- logger = self
- while logger:
- if logger.level:
- return logger.level
- logger = logger.parent
- return NOTSET
日志輸出:
- def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False,
- stacklevel=1):
- ...
- fn, lno, func = "(unknown file)", 0, "(unknown function)"
- ...
- # 生成日志記錄
- record = self.makeRecord(self.name, level, fn, lno, msg, args,
- exc_info, func, extra, sinfo)
- # 使用handler處理日志
- self.handle(record)
日志記錄的生產(chǎn)，就是創(chuàng)建一個LogRecord對象:
- _logRecordFactory = LogRecord
- def makeRecord(self, name, level, fn, lno, msg, args, exc_info,
- func=None, extra=None, sinfo=None):
- ...
- rv = _logRecordFactory(name, level, fn, lno, msg, args, exc_info, func,
- sinfo)
- ...
- return rv
使用logger對象的所有handler處理日志：
- def handle(self, record):
- c = self
- found = 0
- while c:
- for hdlr in c.handlers: # 使用所有的handler處理日志
- found = found + 1
- if record.levelno >= hdlr.level:
- hdlr.handle(record)
root-logger的handler是在config中配置的:
- def basicConfig(**kwargs):
- ...
- root.addHandler(h) # 設(shè)置root的handler
日志記錄對象 LogRecord
日志記錄對象非常簡單:
- class LogRecord(object):
- def __init__(self, name, level, pathname, lineno,
- msg, args, exc_info, func=None, sinfo=None, **kwargs):
- ct = time.time()
- self.name = name # logger名稱
- self.msg = msg # 日志標識信息
- ...
- self.args = args # 變量
- self.levelname = getLevelName(level)
- ...
- def getMessage(self):
- msg = str(self.msg)
- if self.args:
- msg = msg % self.args # 格式化消息
- return msg
處理器對象 Hander
頂級Handler定義了Handler的模版方法
- class Handler(Filterer): # 處理器也可以過濾日志
- def __init__(self, level=NOTSET):
- Filterer.__init__(self)
- self._name = None
- self.level = _checkLevel(level) # handler也有日志級別
- self.formatter = None
- _addHandlerRef(self)
- self.createLock()
- def handle(self, record): # 處理日志
- rv = self.filter(record) # 過濾日志
- if rv:
- self.acquire() # 申請鎖
- try:
- self.emit(record) # 提交記錄，由不同子類實現(xiàn)
- finally:
- self.release() # 釋放鎖
- return rv
默認的console流 StreamHandler
- class StreamHandler(Handler):
- terminator = '\n' # 自動換行
- def __init__(self, stream=None):
- Handler.__init__(self)
- if stream is None:
- stream = sys.stderr # 默認使用stderr輸出
- self.stream = stream
- def emit(self, record):
- try:
- msg = self.format(record) # 格式化日志記錄
- stream = self.stream
- stream.write(msg + self.terminator) # 寫日志
- self.flush() # 刷新寫緩存
- except Exception:
- ...
- def format(self, record):
- if self.formatter:
- fmt = self.formatter
- else:
- fmt = _defaultFormatter
- return fmt.format(record) # 使用格式化器格式化日志記錄
為什么使用stderr，可以看下面的測試中的輸出都是到console:
- print("haha")
- print("fatal error", file=sys.stderr)
- sys.stderr.write("fatal error\n")
格式器對象 Formatter
格式化器主要使用Formatter和Style實現(xiàn)
- class Formatter(object):
- def __init__(self, fmt=None, datefmt=None, style='%', validate=True):
- self._style = _STYLES[style][0](fmt)
- self._fmt = self._style._fmt
- self.datefmt = datefmt
- def format(self, record):
- record.message = record.getMessage()
- s = self.formatMessage(record)
- return s
- def formatMessage(self, record):
- return self._style.format(record) # 格式化
Style類
- class PercentStyle(object):
- default_format = '%(message)s'
- asctime_format = '%(asctime)s'
- asctime_search = '%(asctime)'
- validation_pattern = re.compile(r'%\(\w+\)[#0+ -]*(\*|\d+)?(\.(\*|\d+))?[diouxefgcrsa%]', re.I)
- def __init__(self, fmt):
- self._fmt = fmt or self.default_format
- def usesTime(self):
- return self._fmt.find(self.asctime_search) >= 0
- def validate(self):
- """Validate the input format, ensure it matches the correct style"""
- if not self.validation_pattern.search(self._fmt):
- raise ValueError("Invalid format '%s' for '%s' style" % (self._fmt, self.default_format[0]))
- def _format(self, record):
- return self._fmt % record.__dict__ # 格式化日志記錄對象
- def format(self, record):
- try:
- return self._format(record)
- except KeyError as e:
- raise ValueError('Formatting field not found in record: %s' % e)
滾動日志文件處理器
線上的日志持續(xù)輸出到一個文件的話，會讓文件巨大，即有加劇了丟失的風險，也難以處理。通常有按照大小滾動或者按照日期滾動的方法，這個功能非常重要。先看滾動日志處理器模版:
- class BaseRotatingHandler(logging.FileHandler):
- def emit(self, record):
- try:
- if self.shouldRollover(record): # 判斷是否需要滾動
- self.doRollover() # 滾動日志
- logging.FileHandler.emit(self, record) # 輸出日志
- except Exception:
- self.handleError(record)
- def rotate(self, source, dest):
- if not callable(self.rotator):
- if os.path.exists(source):
- os.rename(source, dest) # 重命名日志文件
- else:
- self.rotator(source, dest)
按大小滾動 RotatingFileHandler
按照文件大小滾動的處理器:
- class RotatingFileHandler(BaseRotatingHandler):
- def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=False):
- if maxBytes > 0:
- mode = 'a'
- BaseRotatingHandler.__init__(self, filename, mode, encoding, delay)
- self.maxBytes = maxBytes # 單個文件大小上限
- self.backupCount = backupCount # 日志備份數(shù)量
- def doRollover(self): # 執(zhí)行滾動
- if self.stream:
- self.stream.close() # 關(guān)閉當前的流
- self.stream = None
- if self.backupCount > 0:
- for i in range(self.backupCount - 1, 0, -1):
- sfn = self.rotation_filename("%s.%d" % (self.baseFilename, i))
- dfn = self.rotation_filename("%s.%d" % (self.baseFilename,
- i + 1))
- if os.path.exists(sfn):
- if os.path.exists(dfn):
- os.remove(dfn)
- os.rename(sfn, dfn)
- dfn = self.rotation_filename(self.baseFilename + ".1")
- if os.path.exists(dfn):
- os.remove(dfn)
- self.rotate(self.baseFilename, dfn) # 重命名文件
- if not self.delay:
- self.stream = self._open() # 如果shouldRollover延遲，可以打開新的流
- def shouldRollover(self, record): # 判斷是否需要滾動
- if self.stream is None: # 立即打開流
- self.stream = self._open()
- if self.maxBytes > 0:
- msg = "%s\n" % self.format(record)
- self.stream.seek(0, 2) #due to non-posix-compliant Windows feature
- if self.stream.tell() + len(msg) >= self.maxBytes: # 判斷大小
- return 1
- return 0
文件大小滾動就是在記錄日志時候判斷文檔是否超過上限，超過則重命名舊日志，生成新日志。
按照日期滾動 TimedRotatingFileHandler
按照日期滾動的處理器:
- class TimedRotatingFileHandler(BaseRotatingHandler):
- def __init__(self, filename, when='h', interval=1, backupCount=0, encoding=None, delay=False, utc=False, atTime=None):
- BaseRotatingHandler.__init__(self, filename, 'a', encoding, delay)
- self.when = when.upper()
- self.backupCount = backupCount
- self.utc = utc
- self.atTime = atTime
- # 日期設(shè)置,支持多種方式
- if self.when == 'S':
- self.interval = 1 # one second
- self.suffix = "%Y-%m-%d_%H-%M-%S"
- self.extMatch = r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}(\.\w+)?$"
- ...
- self.extMatch = re.compile(self.extMatch, re.ASCII)
- self.interval = self.interval * interval # multiply by units requested
- filename = self.baseFilename
- if os.path.exists(filename):
- t = os.stat(filename)[ST_MTIME] # 最后修改時間
- else:
- t = int(time.time())
- self.rolloverAt = self.computeRollover(t) # 提前計算終止時間
- def computeRollover(self, currentTime):
- # 判斷的方法還是很長很復(fù)雜的，先pass
- def shouldRollover(self, record):
- t = int(time.time())
- if t >= self.rolloverAt: # 判斷是否到期
- return 1
- return 0
- def doRollover(self):
- ...
- dfn = self.rotation_filename(self.baseFilename + "." +
- time.strftime(self.suffix, timeTuple))
- # 滾動日志文件
- if os.path.exists(dfn):
- os.remove(dfn)
- self.rotate(self.baseFilename, dfn)
- if self.backupCount > 0:
- for s in self.getFilesToDelete():
- os.remove(s)
- ...
- # 計算下一個時間點
- newRolloverAt = self.computeRollover(currentTime)
- ...
- self.rolloverAt = newRolloverAt
日期滾動就是計算最后時間點，超過時間點則重新生成新的日志文件。
小結(jié)
logging的處理邏輯大概是這樣的:
-
創(chuàng)建Logger對象，提供API，用來接收應(yīng)用程序日志
-
Logger對象包括多個Handler
-
每個Handler有一個Formatter對象
-
每條日志都會生成一個LogRecord對象
-
使用不同的Handler對象將LogRecored對象提交到不同的流
-
每個日志對象通過Formatter格式化輸出
-
可以使用按日期/文件大小的方式進行日志文件的滾動記錄
小技巧
覆蓋對象的 __reduce__ 方法，讓對象支持 reduce 函數(shù):
- class RootLogger(Logger):
- def __init__(self, level):
- Logger.__init__(self, "root", level)
- def __reduce__(self):
- return getLogger, ()
線程鎖的創(chuàng)建和釋放:
- _lock = threading.RLock()
- def _acquireLock():
- if _lock:
- _lock.acquire()
- def _releaseLock():
- if _lock:
- _lock.release()
線程鎖的使用:
- def addHandler(self, hdlr):
- _acquireLock()
- try:
- self.handlers.append(hdlr)
- finally:
- _releaseLock()
- def removeHandler(self, hdlr):
- _acquireLock()
- try:
- self.handlers.remove(hdlr)
- finally:
- _releaseLock()
參考鏈接
-
Logging in Python https://realpython.com/python-logging/
-
日志操作手冊 https://docs.python.org/zh-cn/3.8/howto/logging-cookbook.html#cookbook-rotator-namer
-
Python 的日志記錄工具 https://docs.python.org/zh-cn/3.8/library/logging.html
網(wǎng)頁題目：Python如何僅用5000行代碼，實現(xiàn)強大的logging模塊？
標題鏈接：http://www.5511xx.com/article/djisshe.html
