?Python 如何僅用 5000 行代碼，實現(xiàn)強大的 logging 模塊？
△點擊上方“Python貓”關注 ，回復“2”加入交流群
作者：肖恩
來源：游戲不存在
Python 的 logging 模塊實現(xiàn)了靈活的日志系統(tǒng)。整個模塊僅僅 3 個類，不到 5000 行代碼的樣子，學習它可以加深對程序日志的了解，本文分下面幾個部分:
logging 簡介 logging API 設計 記錄器對象 Logger 日志記錄對象 LogRecord 處理器對象 Hander 格式器對象 Formatter 滾動日志文件處理器 小結 小技巧
logging 簡介
本次代碼使用的是 python 3.8.5 的版本，官方中文文檔 3.8.8 。參考鏈接中官方中文文檔非常詳細，建議先看一遍了解日志使用。
| 類 | 功能 |
|---|---|
| logging-module | logging的API |
| Logger | 日志記錄器對象類，可以創(chuàng)建一個對象用來記錄日志 |
| LogRecord | 日志記錄對象，每條日志記錄都封裝成一個日志記錄對象 |
| Hander | 處理器對象，負責日志輸出到流/文件的控制 |
| Formatter | 格式器，負責日志記錄的格式化 |
| RotatingFileHandler | 按大小滾動的日志文件記錄器 |
| TimedRotatingFileHandler | 按時間滾動的日志文件處理器 |
我們主要研究日志如何輸出到標準窗口這一主線；日志的配置，日志的線程安全及各種特別的Handler等支線可以先忽略。
logging API 設計
先看看日志使用：
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')
輸出內容如下:
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中很重要的設置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) # 設置root的handler
level = kwargs.pop("level", None)
if level is not None:
root.setLevel(level) # 設置日志級別
可以看到，日志的配置主要包括下面幾項:
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] # 獲取已經創(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 # 可以關閉
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)
日志記錄的生產，就是創(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) # 設置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() # 關閉當前的流
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
# 日期設置,支持多種方式
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):
# 判斷的方法還是很長很復雜的，先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
日期滾動就是計算最后時間點，超過時間點則重新生成新的日志文件。
小結
logging的處理邏輯大概是這樣的:
創(chuàng)建Logger對象，提供API，用來接收應用程序日志 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
近期熱門文章推薦：
分享與在看是對我最大的支持！
評論
圖片
表情