Python 如何僅用 5000 行代碼，實現強大的 logging 模塊？

作者：肖恩 2021-03-08 15:04:48

Python 的 logging 模塊實現了靈活的日志系統。整個模塊僅僅 3 個類，不到 5000 行代碼的樣子，學習它可以加深對程序日志的了解

Python 的 logging 模塊實現了靈活的日志系統。整個模塊僅僅 3 個類，不到 5000 行代碼的樣子，學習它可以加深對程序日志的了解，本文分下面幾個部分:

logging 簡介
logging API 設計
記錄器對象 Logger
日志記錄對象 LogRecord
處理器對象 Hander
格式器對象 Formatter
滾動日志文件處理器
小結
小技巧

logging 簡介

本次代碼使用的是 python 3.8.5 的版本，官方中文文檔 3.8.8 。參考鏈接中官方中文文檔非常詳細，建議先看一遍了解日志使用。

類	功能
logging-module	logging的API
Logger	日志記錄器對象類，可以創建一個對象用來記錄日志
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)  # 創建特定的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]  # 獲取已經創建過的同名logger 
            ... 
        else: 
            rv = (self.loggerClass or _loggerClass)(name)  # 創建新的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)

日志記錄的生產，就是創建一個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)  # 提交記錄，由不同子類實現  
            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實現

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)

滾動日志文件處理器

線上的日志持續輸出到一個文件的話，會讓文件巨大，即有加劇了丟失的風險，也難以處理。通常有按照大小滾動或者按照日期滾動的方法，這個功能非常重要。先看滾動日志處理器模版:

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  # 日志備份數量 
         
    def doRollover(self):  # 執行滾動 
        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的處理邏輯大概是這樣的:

創建Logger對象，提供API，用來接收應用程序日志
Logger對象包括多個Handler
每個Handler有一個Formatter對象
每條日志都會生成一個LogRecord對象
使用不同的Handler對象將LogRecored對象提交到不同的流
每個日志對象通過Formatter格式化輸出
可以使用按日期/文件大小的方式進行日志文件的滾動記錄

小技巧

覆蓋對象的 __reduce__ 方法，讓對象支持 reduce 函數:

class RootLogger(Logger): 
    def __init__(self, level): 
        Logger.__init__(self, "root", level) 
 
    def __reduce__(self): 
        return getLogger, ()

線程鎖的創建和釋放:

_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

責任編輯：張燕妮來源：游戲不存在

編程 Python 代碼

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看