如何像专业Python开发者一样使用Python日志
我敢打赌,几乎每个Python开发者在调试代码时都会使用“print”语句。对于原型设计而言,这没有任何问题,但是在实际生产中,有更加有效的方式来处理日志。本文将展示五个实用的理由,为什么Python的“logging”更加灵活和强大,如果您以前没有使用过,那么您绝对应该开始使用。
让我们开始吧。
代码
为了更加实用,让我们考虑一个玩具示例。我创建了一个小应用程序,用于计算两个Python列表的线性回归:
import numpy as npfrom sklearn.linear_model import LinearRegressionfrom typing import List, Optionaldef do_regression(arr_x: List, arr_y: List) -> Optional[List]: """ X和Y列表的线性回归 """ try: x_in = np.array(arr_x).reshape(-1, 1) y_in = np.array(arr_y).reshape(-1, 1) print(f"X: {x_in}") print(f"y: {y_in}") reg = LinearRegression().fit(x_in, y_in) out = reg.predict(x_in) print(f"Out: {out}") print(f"Score: {reg.score(x_in, arr_y)}") print(f"Coef: {reg.coef_}") return out.reshape(-1).tolist() except ValueError as err: print(f"ValueError: {err}") return Noneif __name__ == "__main__": print("App started") ret = do_regression([1,2,3,4], [5,6,7,8]) print(f"LinearRegression result: {ret}")
这段代码可以执行,但是我们能否做得更好?我们显然可以。让我们看看在这段代码中使用” logging ”而不是” print”有哪些优势。
1. 日志级别
让我们对代码进行一些更改:
import loggingdef do_regression(arr_x: List, arr_y: List) -> Optional[List]: """X和Y Python列表的线性回归""" try: x_in = np.array(arr_x).reshape(-1, 1) y_in = np.array(arr_y).reshape(-1, 1) logging.debug(f"X: {x_in}") ... except ValueError as err: logging.error(f"ValueError: {err}") return Noneif __name__ == "__main__": logging.basicConfig(level=logging.DEBUG, format='%(message)s') logging.info("App started") ret = do_regression([1,2,3,4], [5,6,7,8]) logging.info(f"LinearRegression result: {ret}")
在此处,我将“print”调用替换为“logging”调用。我们进行了一些小的更改,但它使输出更加灵活。使用“level”参数,我们现在可以设置不同的日志级别。例如,如果我们使用“level=logging.DEBUG”,那么所有输出都将可见。当我们确信我们的代码已准备好生产时,我们可以将级别更改为“logging.INFO”,并且调试消息将不再显示:
重要的是没有必要更改代码,除了日志本身的初始化!
顺便说一下,所有可用的常量可以在logging/__init__.py文件中找到:
ERROR = 40WARNING = 30INFO = 20DEBUG = 10NOTSET = 0
如我们所见,“ERROR”级别是最高的;通过启用“ERROR”日志级别,我们可以抑制所有其他消息,只显示错误。
2. 格式化
从上一个截图中我们可以看到,控制日志输出很容易。但我们可以做得更多以改进它。我们可以通过提供“格式”字符串来调整输出。例如,我可以指定如下格式:
logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] %(filename)s:%(lineno)d: %(message)s')
没有其他代码更改,我就能看到输出中的时间戳、文件名,甚至行号:
有大约20个不同的参数可用,这些参数可以在手册的“LogRecord属性”段中找到。
3. 将日志保存到文件
Python日志是一个非常灵活的模块,其功能可以很容易地扩展。假设我们想将所有日志保存到文件以供将来分析。为此,我们只需要添加两行代码:
logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] %(message)s', handlers=[logging.FileHandler("debug.log"), logging.StreamHandler()])
正如我们所看到的,我添加了一个新的参数“handlers”。StreamHandler在控制台显示日志,而FileHandler,如其名称所示,将相同的输出保存到文件中。
这个系统确实非常灵活。Python中有许多不同的“handler”对象可用,我鼓励读者自行查阅手册。而且正如我们已经知道的,日志工作几乎是自动的,不需要进一步的代码更改。
4. 旋转日志文件
将日志保存到文件是一个不错的选择,但遗憾的是,磁盘空间是有限的。我们可以通过使用旋转日志文件来轻松解决这个问题:
from logging.handlers import TimedRotatingFileHandler...if __name__ == "__main__": file_handler = TimedRotatingFileHandler( filename="debug.log", when="midnight", interval=1, backupCount=3, ) logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] %(message)s', handlers=[file_handler, logging.StreamHandler()])
所有参数都是自解释的。一个TimedRotatingFileHandler对象将创建一个日志文件,它将在每个午夜更改,仅存储最后三个日志文件。之前的文件将自动重命名为“debug.log.2023.03.03”,在3天间隔后,它们将被删除。
5. 通过套接字发送日志
Python的日志功能非常灵活。如果我们不想将日志保存到本地文件中,我们可以添加一个套接字处理程序,该处理程序将使用特定的IP和端口将日志发送到另一个服务:
from logging.handlers import SocketHandlerlogging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] %(message)s', handlers=[SocketHandler(host="127.0.0.1", port=15001), logging.StreamHandler()])
就是这样,不需要进行任何其他代码更改!
我们还可以创建另一个应用程序,该应用程序将侦听同一端口:
import socketimport loggingimport pickleimport structfrom logging import LogRecordport = 15001stream_handler = logging.StreamHandler()def create_socket() -> socket.socket: """Create the socket""" sock = socket.socket(socket.AF_INET) sock.settimeout(30.0) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) return sockdef read_socket_data(conn_in: socket.socket): """Read data from socket""" while True: data = conn_in.recv(4) # Data: 4 bytes length + body if len(data) > 0: body_len = struct.unpack(">L", data)[0] data = conn_in.recv(body_len) record: LogRecord = logging.makeLogRecord(pickle.loads(data)) stream_handler.emit(record) else: logging.debug("Socket connection lost") returnif __name__ == "__main__": logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] %(message)s', handlers=[stream_handler]) sock = create_socket() sock.bind(("127.0.0.1", port)) # 仅本地连接 sock.listen(1) # 一个客户端可以连接 logging.debug("Logs listening thread started") while True: try: conn, _ = sock.accept() logging.debug("Socket connection established") read_socket_data(conn) except socket.timeout: logging.debug("Socket listening: no data")
关键在于使用emit方法,该方法将套接字接收到的所有远程数据添加到活动的StreamHandler中。
6. 奖励:日志过滤器
最后,对于那些足够细心地阅读到这一部分的读者,我们有个小小的奖励。添加自定义过滤器到日志中也很容易。假设我们只想记录X和Y值以供将来分析。我们可以很容易地创建一个新的Filter类,该类仅保存包含“x:”或“y:”记录的字符串:
from logging import LogRecord, Filterclass DataFilter(Filter): """用于记录消息的过滤器""" def filter(self, record: LogRecord) -> bool: """仅保存筛选后的数据""" return "x:" in record.msg.lower() or "y:" in record.msg.lower()
然后,我们可以很容易地将此过滤器添加到文件日志中。我们的控制台输出将保持不变,但文件中将仅包含“x:”和“y:”值。
file_handler = logging.FileHandler("debug.log")file_handler.addFilter(DataFilter())logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] %(message)s', handlers=[file_handler, logging.StreamHandler()])
结论
在这篇简短的文章中,我们学习了几种将日志纳入Python应用程序的简单方法。Python中的日志记录是一个非常灵活的框架,绝对值得花些时间研究它的工作原理。
感谢您的阅读,祝您在未来的实验中好运。
如果您喜欢这个故事,请随时订阅小猪AI,您将获得我的新文章发布通知,以及其他作者的数千个故事的完整访问权限。