Loguru: 更優(yōu)雅的日志記錄解決方案
loguru 是一個(gè) Python 簡(jiǎn)易且強(qiáng)大的第三方日志記錄庫(kù)，該庫(kù)旨在通過(guò)添加一系列有用的功能來(lái)解決標(biāo)準(zhǔn)記錄器的注意事項(xiàng)，從而減少 Python 日志記錄的痛苦。
Github 項(xiàng)目地址：https://github.com/Delgan/loguru
Start：10.5k
1. 引入原因
簡(jiǎn)單且方便的幫助我們輸出需要的日志信息
使用 Python 來(lái)寫程序或者腳本的話，常常遇到的問(wèn)題就是需要對(duì)日志進(jìn)行刪除。一方面可以幫助我們?cè)诔绦虺鰡?wèn)題的時(shí)候排除問(wèn)題，二來(lái)可以幫助我們記錄需要關(guān)注的信息。
但是，使用自帶自帶的 logging 模塊的話，則需要我們進(jìn)行不同的初始化等相關(guān)工作。對(duì)應(yīng)不熟悉該模塊的同學(xué)來(lái)說(shuō)，還是有些費(fèi)勁的，比如需要配置 Handler/Formatter 等。
import logging
logger = logging.getLogger('xxx')
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(name)-12s %(levelname)-8s %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
logger.debug('This is a %s', 'test')
而 loguru 就是一個(gè)可以 開箱即用 的日志記錄模塊，我們不再需要復(fù)雜的初始化操作就可以通過(guò)如下命令來(lái)記錄日志信息了。
# pip $ pip install loguru
2. 功能特性
有很多優(yōu)點(diǎn)，以下列舉了其中比較重要的幾點(diǎn)：
開箱即用，無(wú)需準(zhǔn)備
無(wú)需初始化，導(dǎo)入函數(shù)即可使用
更容易的文件日志記錄與轉(zhuǎn)存/保留/壓縮方式
更優(yōu)雅的字符串格式化輸出
可以在線程或主線程中捕獲異常
可以設(shè)置不同級(jí)別的日志記錄樣式
支持異步，且線程和多進(jìn)程安全
支持惰性計(jì)算
適用于腳本和庫(kù)
完全兼容標(biāo)準(zhǔn)日志記錄
更好的日期時(shí)間處理
3. 快速上手
loguru 的常用操作和功能：
[1] 開箱即用，無(wú)需準(zhǔn)備
loguru 并沒有什么黑科技，只是它預(yù)先幫助我們?cè)O(shè)置好了相關(guān)的配置，我們導(dǎo)入之后即可直接使用。
from loguru import logger
logger.debug("That's it, beautiful and simple logging!")
[2] 無(wú)需初始化，導(dǎo)入函數(shù)即可使用
如何添加處理程序(handler)呢？
如何設(shè)置日志格式(logs formatting)呢？
如何過(guò)濾消息(filter messages)呢？
如何如何設(shè)置級(jí)別(log level)呢？
# add
logger.add(sys.stderr, \
format="{time} {level} {message}",\
filter="my_module",\
level="INFO")
[3] 更容易的文件日志記錄與轉(zhuǎn)存/保留/壓縮方式
# 日志文件記錄
logger.add("file_{time}.log")
# 日志文件轉(zhuǎn)存
logger.add("file_{time}.log", rotation="500 MB")
logger.add("file_{time}.log", rotation="12:00")
logger.add("file_{time}.log", rotation="1 week")
# 多次時(shí)間之后清理
logger.add("file_X.log", retention="10 days")
# 使用zip文件格式保存
logger.add("file_Y.log", compression="zip")
[4] 更優(yōu)雅的字符串格式化輸出
logger.info(
"If you're using Python {}, prefer {feature} of course!",
3.6, feature="f-strings")
[5] 在線程或主線程中捕獲異常
@logger.catch def my_function(x, y, z): # An error? It's caught anyway! return 1 / (x + y + z) my_function(0, 0, 0)
[6] 可以設(shè)置不同級(jí)別的日志記錄樣式
Loguru 會(huì)自動(dòng)為不同的日志級(jí)別，添加不同的顏色進(jìn)行區(qū)分，當(dāng)然我們也是可以自定義自己喜歡的顯示顏色樣式的。
logger.add(sys.stdout,
colorize=True,
format="<green>{time}</green> <level>{message}</level>")
logger.add('logs/z_{time}.log',
level='DEBUG',
format='{time:YYYY-MM-DD :mm:ss} - {level} - {file} - {line} - {message}',
rotation="10 MB")
[7] 支持異步且線程和多進(jìn)程安全
默認(rèn)情況下，添加到 logger 中的日志信息都是線程安全的。但這并不是多進(jìn)程安全的，我們可以通過(guò)添加 enqueue 參數(shù)來(lái)確保日志完整性。
如果我們想要在異步任務(wù)中使用日志記錄的話，也是可以使用同樣的參數(shù)來(lái)保證的。并且通過(guò) complete() 來(lái)等待執(zhí)行完成。
# 異步寫入
logger.add("some_file.log", enqueue=True)
[8] 異常的完整性描述
用于記錄代碼中發(fā)生的異常的 bug 跟蹤，Loguru 通過(guò)允許顯示整個(gè)堆棧跟蹤(包括變量值)來(lái)幫助您識(shí)別問(wèn)題。
logger.add("out.log", backtrace=True, diagnose=True)
def func(a, b):
return a / b
def nested(c):
try:
func(5, c)
except ZeroDivisionError:
logger.exception("What?!")
nested(0)
[9] 結(jié)構(gòu)化日志記錄
對(duì)日志進(jìn)行序列化以便更容易地解析或傳遞數(shù)據(jù)結(jié)構(gòu)，使用序列化參數(shù)，在將每個(gè)日志消息發(fā)送到配置的接收器之前，將其轉(zhuǎn)換為 JSON 字符串。
同時(shí)，使用 bind() 方法，可以通過(guò)修改額外的 record 屬性來(lái)將日志記錄器消息置于上下文中。還可以通過(guò)組合 bind() 和 filter 對(duì)日志進(jìn)行更細(xì)粒度的控制。
最后 patch() 方法允許將動(dòng)態(tài)值附加到每個(gè)新消息的記錄 dict 上。
# 序列化為json格式
logger.add(custom_sink_function, serialize=True)
# bind方法的用處
logger.add("file.log", format="{extra[ip]} {extra[user]} {message}")
context_logger = logger.bind(ip="192.168.0.1", user="someone")
context_logger.info("Contextualize your logger easily")
context_logger.bind(user="someone_else").info("Inline binding of extra attribute")
context_logger.info("Use kwargs to add context during formatting: {user}", user="anybody")
# 粒度控制
logger.add("special.log", filter=lambda record: "special" in record["extra"])
logger.debug("This message is not logged to the file")
logger.bind(special=True).info("This message, though, is logged to the file!")
# patch()方法的用處
logger.add(sys.stderr, format="{extra[utc]} {message}")
logger = logger.patch(lambda record: record["extra"].update(utc=datetime.utcnow()))
[10] 惰性計(jì)算
有時(shí)希望在生產(chǎn)環(huán)境中記錄詳細(xì)信息而不會(huì)影響性能，可以使用 opt() 方法來(lái)實(shí)現(xiàn)這一點(diǎn)。
logger.opt(lazy=True).debug("If sink level <= DEBUG: {x}", x=lambda: expensive_function(2**64))
# By the way, "opt()" serves many usages
logger.opt(exception=True).info("Error stacktrace added to the log message (tuple accepted too)")
logger.opt(colors=True).info("Per message <blue>colors</blue>")
logger.opt(record=True).info("Display values from the record (eg. {record[thread]})")
logger.opt(raw=True).info("Bypass sink formatting\n")
logger.opt(depth=1).info("Use parent stack context (useful within wrapped functions)")
logger.opt(capture=False).info("Keyword arguments not added to {dest} dict", dest="extra")
[11] 可定制的級(jí)別
new_level = logger.level("SNAKY", no=38, color="<yellow>", icon="🐍")
logger.log("SNAKY", "Here we go!")
[12] 適用于腳本和庫(kù)
# For scripts
config = {
"handlers": [
{"sink": sys.stdout, "format": "{time} - {message}"},
{"sink": "file.log", "serialize": True},
],
"extra": {"user": "someone"}
}
logger.configure(**config)
# For libraries
logger.disable("my_library")
logger.info("No matter added sinks, this message is not displayed")
logger.enable("my_library")
logger.info("This message however is propagated to the sinks")
[13] 完全兼容標(biāo)準(zhǔn)日志記錄
希望使用 Loguru 作為內(nèi)置的日志處理程序？
需要將 Loguru 消息到標(biāo)準(zhǔn)日志？
想要攔截標(biāo)準(zhǔn)的日志消息到 Loguru 中匯總？
handler = logging.handlers.SysLogHandler(address=('localhost', 514))
logger.add(handler)
class PropagateHandler(logging.Handler):
def emit(self, record):
logging.getLogger(record.name).handle(record)
logger.add(PropagateHandler(), format="{message}")
class InterceptHandler(logging.Handler):
def emit(self, record):
# Get corresponding Loguru level if it exists
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message
frame, depth = logging.currentframe(), 2
while frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
logging.basicConfig(handlers=[InterceptHandler()], level=0)
[14] 方便的解析器
從生成的日志中提取特定的信息通常很有用，這就是為什么 Loguru 提供了一個(gè) parse() 方法來(lái)幫助處理日志和正則表達(dá)式。
pattern = r"(?P<time>.*) - (?P<level>[0-9]+) - (?P<message>.*)" # Regex with named groups
caster_dict = dict(time=dateutil.parser.parse, level=int) # Transform matching groups
for groups in logger.parse("file.log", pattern, cast=caster_dict):
print("Parsed:", groups)
# {"level": 30, "message": "Log example", "time": datetime(2018, 12, 09, 11, 23, 55)}
[15] 通知機(jī)制
import notifiers
params = {
"username": "you@gmail.com",
"password": "abc123",
"to": "dest@gmail.com"
}
# Send a single notification
notifier = notifiers.get_notifier("gmail")
notifier.notify(message="The application is running!", **params)
# Be alerted on each error message
from notifiers.logging import NotificationHandler
handler = NotificationHandler("gmail", defaults=params)
logger.add(handler, level="ERROR")
[16] Flask 框架集成
現(xiàn)在最關(guān)鍵的一個(gè)問(wèn)題是如何兼容別的 logger，比如說(shuō) tornado 或者 django 有一些默認(rèn)的 logger。
經(jīng)過(guò)研究，最好的解決方案是參考官方文檔的，完全整合 logging 的工作方式。比如下面將所有的 logging都用 loguru 的 logger 再發(fā)送一遍消息。
import logging
import sys
from pathlib import Path
from flask import Flask
from loguru import logger
app = Flask(__name__)
class InterceptHandler(logging.Handler):
def emit(self, record):
logger_opt = logger.opt(depth=6, exception=record.exc_info)
logger_opt.log(record.levelname, record.getMessage())
def configure_logging(flask_app: Flask):
"""配置日志"""
path = Path(flask_app.config['LOG_PATH'])
if not path.exists():
path.mkdir(parents=True)
log_name = Path(path, 'sips.log')
logging.basicConfig(handlers=[InterceptHandler(level='INFO')], level='INFO')
# 配置日志到標(biāo)準(zhǔn)輸出流
logger.configure(handlers=[{"sink": sys.stderr, "level": 'INFO'}])
# 配置日志到輸出到文件
logger.add(log_name, rotation="500 MB", encoding='utf-8', colorize=False, level='INFO')
4. 要點(diǎn)解析
介紹主要函數(shù)的使用方法和細(xì)節(jié) - add()的創(chuàng)建和刪除
add() - 非常重要的參數(shù) sink 參數(shù)
具體的實(shí)現(xiàn)規(guī)范可以參見官方文檔
可以實(shí)現(xiàn)自定義 Handler 的配置，比如 FileHandler、StreamHandler 等等
可以自行定義輸出實(shí)現(xiàn)
代表文件路徑，會(huì)自動(dòng)創(chuàng)建對(duì)應(yīng)路徑的日志文件并將日志輸出進(jìn)去
例如 sys.stderr 或者 open('file.log', 'w') 都可以
可以傳入一個(gè) file 對(duì)象
可以直接傳入一個(gè) str 字符串或者 pathlib.Path 對(duì)象
可以是一個(gè)方法
可以是一個(gè) logging 模塊的 Handler
可以是一個(gè)自定義的類
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 ):
另外添加 sink 之后我們也可以對(duì)其進(jìn)行刪除，相當(dāng)于重新刷新并寫入新的內(nèi)容。刪除的時(shí)候根據(jù)剛剛 add 方法返回的 id 進(jìn)行刪除即可?？梢园l(fā)現(xiàn)，在調(diào)用 remove 方法之后，確實(shí)將歷史 log 刪除了。但實(shí)際上這并不是刪除，只不過(guò)是將 sink 對(duì)象移除之后，在這之前的內(nèi)容不會(huì)再輸出到日志中，這樣我們就可以實(shí)現(xiàn)日志的刷新重新寫入操作。
from loguru import logger
trace = logger.add('runtime.log')
logger.debug('this is a debug message')
logger.remove(trace)
logger.debug('this is another debug message')
