Python 日志管理模塊loguru


Python中的日志管理模塊可以使用自帶的logging,也可使用第三方Loguru模塊,使用logging需要配置Handler、Formatter 進行一些處理,配置比較繁瑣,

而使用Loguru則較為簡單。

安裝

pip install loguru

基本使用

loguru庫的使用可以說是十分簡單,我們直接可以通過導入它本身封裝好的logger 類就可以直接進行調用。

logger本身就是一個已經實例化好的對象,如果沒有特殊的配置需求,那么自身就已經帶有通用的配置參數;同時它的用法和logging庫輸出日志時的用法一致。

from loguru import logger

logger.debug('this is a debug message')
logger.info('this is info message')
logger.warning('this is warning message')
logger.error('this is error message')
logger.info('this is info message')
logger.success('this is success message!')
logger.critical('this is critical message!')

執行結果 :

從loguru庫引入logger后,直接調用其 info,debug,error 方法即可。可以看到其默認的輸出格式是上面的內容,有時間、級別、模塊名、行號以及日志信息,不需要手動創建 logger,直接使用即可。上面的日志信息是直接輸出在控制台的,如果想要輸出到其他的位置,比如存為文件,我們只需要使用一行代碼聲明即可。例如將結果同時輸出到一個 runtime.log 文件里面,可以這么寫:

from loguru import logger

logger.add('runtime.log')
logger.debug('this is a debug message')
logger.info('this is info message')
logger.warning('this is warning message')
logger.error('this is error message')
logger.info('this is info message')
logger.success('this is success message!')
logger.critical('this is critical message!')

執行結果:

運行之后會發現目錄下 my_log.log 出現了剛剛控制台輸出的 DEBUG 信息。上面只是基礎用法,更詳細的在下面。

詳細使用

loguru 對輸出到文件的配置有非常強大的支持,比如支持輸出到多個文件,分級別分別輸出,過大創建新文件,過久自動刪除等等。

(一)add 方法定義

def add(
        self,
        sink,
        *,
        level=_defaults.LOGURU_LEVEL,
        format=_defaults.LOGURU_FORMAT,
        filter=_defaults.LOGURU_FILTER,
        colorize=_defaults.LOGURU_COLORIZE,
        serialize=_defaults.LOGURU_SERIALIZE,
        backtrace=_defaults.LOGURU_BACKTRACE,
        diagnose=_defaults.LOGURU_DIAGNOSE,
        enqueue=_defaults.LOGURU_ENQUEUE,
        catch=_defaults.LOGURU_CATCH,
        **kwargs
    ):
    pass

skin參數可以傳入多種不同的數據結構,如下:

  • sink 可以傳入一個 file 對象,例如 sys.stderr 或者 open('file.log', 'w') 都可以。
  • sink 可以直接傳入一個 str 字符串或者 pathlib.Path 對象,其實就是代表文件路徑的,如果識別到是這種類型,它會自動創建對應路徑的日志文件並將日志輸出進去。
  • sink 可以是一個方法,可以自行定義輸出實現。
  • sink 可以是一個 logging 模塊的 Handler,比如 FileHandler、StreamHandler 等等,或者上文中我們提到的 CMRESHandler 照樣也是可以的,這樣就可以實現自定義 Handler 的配置。
  • sink 還可以是一個自定義的類,具體的實現規范可以參見官方文檔。https://loguru.readthedocs.io/en/stable/api/logger.html#sink

(二) 基本參數

其他參數例如 format、filter、level 等等。其實它們的概念和格式和 logging 模塊都是基本一樣的了,例如這里使用 format、filter、level 來規定輸出的格式:

logger.add('runtime.log', format="{time} {level} {message}", filter="my_module", level="INFO",encoding="utf-8")

(三)刪除 sink

添加 sink 之后可以對其進行刪除,相當於重新刷新並寫入新的內容。刪除的時候根據剛剛 add 方法返回的 id 進行刪除即可,看下面的例子:

from loguru import logger

trace = logger.add('my_log.log')
logger.debug('this is a debug message')
logger.remove(trace)
logger.debug('this is another debug message')

看這里,我們首先 add了一個sink,然后獲取它的返回值,賦值為tarce。隨后輸出了一條日志,然后將trace變量傳給remove方法,再次輸出一條日志,看看結果是怎樣的。

控制台輸出如下:

日志文件 my_log.log 內容如下:

 (四)rotation 配置

用了 loguru 我們還可以非常方便地使用rotation配置,比如我們想一天輸出一個日志文件,或者文件太大了自動分隔日志文件,我們可以直接使用add方法的rotation參數進行配置。

logger.add('runtime_{time}.log', rotation="500 MB")  # log文件超過500M時會新建一個log文件

通過這樣的配置我們就可以實現每 500MB 存儲一個文件,每個 log 文件過大就會新創建一個 log 文件。我們在配置 log 名字時加上了一個 time 占位符,這樣在生成時可以自動將時間替換進去,生成一個文件名包含時間的 log 文件。

使用 rotation 參數實現定時創建 log 文件

logger.add('runtime_{time}.log', rotation='00:00') # 每天0點新建一個log文件

另外我們也可以配置 log 文件的循環時間,比如每隔一周創建一個 log 文件,寫法如下:

logger.add('runtime_{time}.log', rotation='1 week') #每隔一周創建一個 log 文件

(五) retention 配置

很多情況下,一些非常久遠的 log 對我們來說並沒有什么用處了,它白白占據了一些存儲空間,不清除掉就會非常浪費。retention 這個參數可以配置日志的最長保留時間。

retention 這個參數可以配置日志的最長保留時間。

logger.add('runtime.log', retention='10 days') # 保留最新10天的log

(六)compression 配置

loguru 還可以配置文件的壓縮格式,比如使用 zip 文件格式保存,示例如下:

logger.add('runtime.log', compression='zip') # 使用zip文件格式保存

(七)serialize 序列化

如果在實際中你不太喜歡以文件的形式保留日志,那么你也可以通過serialize參數將其轉化成序列化的json格式,最后將導入類似於MongoDB、ElasticSearch 這類數NoSQL 數據庫中用作后續的日志分析。

from loguru import logger
import os

logger.add(os.path.expanduser("~/Desktop/testlog.log"), serialize=True)
logger.info("hello, world!")

最后保存的日志都是序列化后的單條記錄:

{
    "text": "2022-4-19 22:59:36.902 | INFO     | __main__:<module>:6 - hello, world\n",
    "record": {
        "elapsed": {
            "repr": "0:00:00.005412",
            "seconds": 0.005412
        },
        "exception": null,
        "extra": {},
        "file": {
            "name": "log_test.py",
            "path": "/Users/Bobot/PycharmProjects/docs-python/src/loguru/log_test.py"
        },
        "function": "<module>",
        "level": {
            "icon": "\u2139\ufe0f",
            "name": "INFO",
            "no": 20
        },
        "line": 6,
        "message": "hello, world",
        "module": "log_test",
        "name": "__main__",
        "process": {
            "id": 12662,
            "name": "MainProcess"
        },
        "thread": {
            "id": 4578131392,
            "name": "MainThread"
        },
        "time": {
            "repr": "2022-4-19 22:59:36.902358+08:00",
            "timestamp": 1602066216.902358
        }
    }
}

(八)字符串格式化

loguru 在輸出 log 的時候還提供了非常友好的字符串格式化功能,像這樣:

logger.info('If you are using Python {}, prefer {feature} of course!', 3.6, feature='f-strings')

 (九)Traceback 記錄

在很多情況下,如果遇到運行錯誤,而我們在打印輸出 log 的時候萬一不小心沒有配置好 Traceback 的輸出,很有可能我們就沒法追蹤錯誤所在了。

但用了 loguru 之后,我們用它提供的裝飾器就可以直接進行 Traceback 的記錄,類似這樣的配置即可:

@logger.catch
def my_function(x, y, z):
    # An error? It's caught anyway!
    return 1 / (x + y + z)

我們做個測試,我們在調用時三個參數都傳入 0,直接引發除以 0 的錯誤,看看會出現什么情況:

my_function(0, 0, 0)

運行完畢之后,可以發現 log 里面就出現了 Traceback 信息,而且給我們輸出了當時的變量值,真的是不能再贊了!結果如下:

因此,用 loguru 可以非常方便地實現日志追蹤,debug 效率可能要高上十倍了?

(十)與 Logging 完全兼容(Entirely Compatible)

盡管說loguru算是重新「造輪子」,但是它也能和logging庫很好地兼容。到現在我們才談論到dd()方法的第一個參數sink。

這個參數的英文單詞動詞有「下沉、浸沒」等意,對於外國人來說在理解上可能沒什么難的,可對我們國人來說,這可之前logging庫中的handler概念還不好理解。好在前面我有說過,loguru和logging庫的使用上存在相似之處,因此在后續的使用中其實我們就可以將其理解為 handler,只不過它的范圍更廣一些,可以除了 handler之外的字符串、可調用方法、協程對象等。

loguru 官方文檔對這一參數的解釋是:

 object in charge of receiving formatted logging messages and propagating them to an appropriate endpoint.

翻譯過來就是「一個用於接收格式化日志信息並將其傳輸合適端點的對象」,進一步形象理解就像是一個「分流器」.

import logging.handlers
import os
import sys

from loguru import logger

LOG_FILE = os.path.expanduser("~/Desktop/testlog.log")
file_handler = logging.handlers.RotatingFileHandler(LOG_FILE, encoding="utf-8")
logger.add(file_handler)
logger.debug("hello, world")

當然目前只是想在之前基於logging寫好的模塊中集成loguru,只要重新編寫一個繼承自logging.handler 類並實現了emit()方法的Handler即可。

比如flask項目中,直接調用setup_loguru()就可以了

def setup_loguru(app,log_level='WARNING'):
    logger.add(
        'logs/{time:%Y-%m-%d}.log',
        level='DEBUG',
        format='{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}',
        backtrace=False,
        rotation='00:00',
        retention='20 days',
        encoding='utf-8'
    )

    app.logger.addHandler(InterceptHandler())
    logging.basicConfig(handlers=[InterceptHandler()], level=log_level)

(十一)Loguru模塊下pytest和HTMLTestRunner生成的html沒有顯示日志內容 

最近將原有logging日志系統替換成了loguru,loguru的好處不用多說,簡單好用。配置起來也比lgging方便多了。封裝代碼如下

import time, os
from loguru import logger

LOG_DIR = os.path.abspath(os.path.dirname(__file__)) # 日志保存路徑

class Log:
    """輸出日志到文件和控制台"""
    def __init__(self):
        # 文件的命名
        log_name = f"test_{time.strftime('%Y-%m-%d', time.localtime()).replace('-','_')}.log"
        log_path = os.path.join(LOG_DIR, log_name)
        # 判斷日志文件夾是否存在,不存則創建
        if not os.path.exists(LOG_DIR):
            os.mkdir(LOG_DIR)
        # 日志輸出格式
        formatter = "{time:YYYY-MM-DD HH:mm:ss} | {level}: {message}"
        # 日志寫入文件
        logger.add(log_path,   # 寫入目錄指定文件
               format=formatter,
               encoding='utf-8',
               retention='10 days',  # 設置歷史保留時長
               backtrace=True,  # 回溯
               diagnose=True,   # 診斷
               enqueue=True,   # 異步寫入
               # rotation="5kb",  # 切割,設置文件大小,rotation="12:00",rotation="1 week"
               # filter="my_module"  # 過濾模塊
               # compression="zip"   # 文件壓縮
            )

    def debug(self, msg):
        logger.debug(msg)

    def info(self, msg):
        logger.info(msg)

    def warning(self, msg):
        logger.warning(msg)

    def error(self, msg):
        logger.error(msg)

log = Log()

將上述封裝代碼引入,生成的html仍沒有日志。點擊通過無法查看日志。

原來使用HTMLTestRunner生成html測試報告時,報告中只有console控制台上輸出,logging的輸出無法保存,如果要在報告中加入每一個測試用例執行的日志信息,則需要改HTMLTestRunner源碼。

修改_TestResult類,同時別忘了在文件最上面import logging。

import logging
class _TestResult(TestResult):
    # note: _TestResult is a pure representation of results.
    # It lacks the output and reporting ability compares to unittest._TextTestResult.

    def __init__(self, verbosity=1, retry=0,save_last_try=False):
        TestResult.__init__(self)

        self.stdout0 = None
        self.stderr0 = None
        self.success_count = 0
        self.failure_count = 0
        self.error_count = 0
        self.skip_count = 0
        self.verbosity = verbosity
        # result is a list of result in 4 tuple
        # (
        #   result code (0: success; 1: fail; 2: error;3:skip),
        #   TestCase object,
        #   Test output (byte string),
        #   stack trace,
        # )
        self.result = []
        self.retry = retry
        self.trys = 0
        self.status = 0

        self.save_last_try = save_last_try
        self.outputBuffer = StringIO.StringIO()
        self.logger = logging.getLogger() # 新增這一行

startTest函數中初始化logging.Handler

def startTest(self, test):
        # test.imgs = []
        test.imgs = getattr(test, "imgs", [])
        # TestResult.startTest(self, test)
        self.outputBuffer.seek(0)
        self.outputBuffer.truncate()
        stdout_redirector.fp = self.outputBuffer
        stderr_redirector.fp = self.outputBuffer
        self.stdout0 = sys.stdout
        self.stderr0 = sys.stderr
        sys.stdout = stdout_redirector
        sys.stderr = stderr_redirector
        
        # 新增如下代碼
        self.log_cap = StringIO.StringIO()
        self.ch = logging.StreamHandler(self.log_cap)
        self.ch.setLevel(logging.DEBUG)
        formatter = logging.Formatter('[%(levelname)s][%(asctime)s] [%(filename)s]->[%(funcName)s] line:%(lineno)d ---> %(message)s')
        self.ch.setFormatter(formatter)
        self.logger.addHandler(self.ch)

complete_output函數的返回值中加入logging存在內存中的輸出,用換行符隔開

def complete_output(self):
        """
        Disconnect output redirection and return buffer.
        Safe to call multiple times.
        """
        if self.stdout0:
            sys.stdout = self.stdout0
            sys.stderr = self.stderr0
            self.stdout0 = None
            self.stderr0 = None
        # return self.outputBuffer.getvalue()
        return self.outputBuffer.getvalue()+'\n'+self.log_cap.getvalue() # 新增這一行

每個用例執行完后,清除handler,修改stopTest函數

def stopTest(self, test):
        # Usually one of addSuccess, addError or addFailure would have been called.
        # But there are some path in unittest that would bypass this.
        # We must disconnect stdout in stopTest(), which is guaranteed to be called.
        if self.retry and self.retry>=1:
            if self.status == 1:
                self.trys += 1
                if self.trys <= self.retry:
                    if self.save_last_try:
                        t = self.result.pop(-1)
                        if t[0]==1:
                            self.failure_count -=1
                        else:
                            self.error_count -= 1
                    test=copy.copy(test)
                    sys.stderr.write("Retesting... ")
                    sys.stderr.write(str(test))
                    sys.stderr.write('..%d \n' % self.trys)
                    doc = getattr(test,'_testMethodDoc',u"") or u''
                    if doc.find('_retry')!=-1:
                        doc = doc[:doc.find('_retry')]
                    desc ="%s_retry:%d" %(doc, self.trys)
                    if not PY3K:
                        if isinstance(desc, str):
                            desc = desc.decode("utf-8")
                    test._testMethodDoc = desc
                    test(self)
                else:
                    self.status = 0
                    self.trys = 0
        #self.complete_output()
        a = self.complete_output()
        # 清除log的handle,新增如下代碼
        self.logger.removeHandler(self.ch)
        return a

可能各自用的HTMLTestRunner版本內容不一樣,均只需按照上述修改即可。

按照上述修改,再次運行,生成的html報告還是沒有日志內容,會不會是loguru和logging不兼容?

import time, os, logging
from loguru import logger
from settings import LOG_DIR # 日志保存路徑

# 新增如下三行代碼
class PropogateHandler(logging.Handler):
    def emit(self, record):
        logging.getLogger(record.name).handle(record)
        
class Log:
    """輸出日志到文件和控制台"""
    def __init__(self):
        # 文件的命名
        log_name = f"test_{time.strftime('%Y-%m-%d', time.localtime()).replace('-','_')}.log"
        log_path = os.path.join(LOG_DIR, log_name)
        # 判斷日志文件夾是否存在,不存則創建
        if not os.path.exists(LOG_DIR): 
            os.mkdir(LOG_DIR)
        # 日志輸出格式
        formatter = "{time:YYYY-MM-DD HH:mm:ss} | {level}: {message}"
        # 日志寫入文件
        logger.add(log_path,   # 寫入目錄指定文件
               format=formatter,
               encoding='utf-8',   
               retention='10 days',  # 設置歷史保留時長
               backtrace=True,  # 回溯
               diagnose=True,   # 診斷
               enqueue=True,   # 異步寫入
               # rotation="5kb",  # 切割,設置文件大小,rotation="12:00",rotation="1 week"
               # filter="my_module"  # 過濾模塊
               # compression="zip"   # 文件壓縮
              ) 
        # 新增代碼
        logger.add(PropogateHandler(), format=formatter)

    def debug(self, msg):
        logger.debug(msg)

    def info(self, msg):
        logger.info(msg)

    def warning(self, msg):
        logger.warning(msg)

    def error(self, msg):
        logger.error(msg) 

log = Log()

loguru二次封裝,結合裝飾器獲取日志:

from loguru import logger
import datetime
import os,sys
from functools import wraps


class MyLoggers(object):
    def __init__(self,log_filename=None,is_stdout=True,need_write_log=True):
        self._logger = logger
        # 判斷是否添加控制台輸出的格式, sys.stdout為輸出到屏幕;
        if is_stdout:
            # 清空所有設置
            self._logger.remove()
            self._logger.add(
                sink=sys.stdout,
                format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "  # 顏色>時間
                       "{process.name} | "  # 進程名
                       "{thread.name} | "  # 線程名
                       "<cyan>{module}</cyan>.<cyan>{function}</cyan>"  # 模塊名.方法名
                       ":<cyan>{line}</cyan> | "  # 行號
                       "<level>{level}</level>: "  # 等級
                       "<level>{message}</level>",  # 日志內容
            )
        # 判斷是否需要寫入logger日志文件
        if need_write_log:
            # self._logger.remove(handler_id=None) #只寫入文件中,不輸入控制台
            self._logger.add(
                # 水槽,分流器,可以用來輸入路徑
                sink= self.get_log_path(log_filename=log_filename),
                # 日志創建周期
                rotation='00:00',
                # 保存
                retention='1 days',
                # 文件的壓縮格式
                compression='zip',
                # 編碼格式
                encoding="utf-8",
                # 具有使日志記錄調用非阻塞的優點(適用於多線程)
                enqueue=True,
                # 日志級別
                level='INFO',
                # 時間|進程名|線程名|模塊|方法|行號|日志等級|日志信息
                format="{time:YYYY-MM-DD HH:mm:ss} | {process.name} | {thread.name} | {module}.{function}:{line} | {level}:{message}"
            )

    def get_log_path(self,log_filename=None):
        # 項目日志目錄
        project_log_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'log'))
        if not os.path.exists(project_log_dir):
            os.makedirs(project_log_dir)
        # 日志文件名
        if log_filename is not None:
            project_log_filename = '{}.log'.format(log_filename)
        else:
            project_log_filename = 'runtime_{}.log'.format(datetime.date.today())
        # 日志文件路徑
        project_log_path = os.path.join(project_log_dir, project_log_filename)
        return project_log_path

    def info(self,mesg):
        self._logger.info(mesg)

    def debug(self,mesg):
        self._logger.debug(mesg)

    def warning(self,mesg):
        self._logger.warning(mesg)

    def error(self,mesg):
        self._logger.error(mesg)


    def get_logger(self):
        return self._logger

my_loger = MyLoggers(log_filename='mylog', need_write_log=True, is_stdout=True).get_logger()
# 裝飾器用來撲獲日志(不帶參數)
def wrapper_log(func):
    @wraps(func)
    def inner(*args,**kwargs):
        my_loger.info(f'{func.__name__}')
        try:
            func(*args,**kwargs)
        except Exception as error_mesg:
            print(error_mesg)
            my_loger.exception(error_mesg)
    return inner()

# 裝飾器用來撲獲日志(帶參數)
def loger(logger_object):
    def wrapper(func):
        def inner(*args,**kwargs):
            logger_object.info(f'{func.__name__}')
            try:
                func(*args,**kwargs)
            except Exception as error_mesg:
                logger_object.exception(error_mesg)
        return inner()
    return wrapper



@wrapper_log
def test_func(a,b):
    print(a/b)

@loger(my_loger)
def test_demo(a,b):
    return a/b

test_func(1/0)
test_demo(1/0)

 

原文來源於:https://blog.51cto.com/u_15127517/4403171

另外loguru還有很多很多強大的功能,這里就不再一一展開講解了,更多的內容大家可以看看 loguru的官方文檔詳細了解一下:

https://loguru.readthedocs.io/en/stable/index.html

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM