IT评测·应用市场-qidao123.com技术社区

标题: [flask]自定义请求日志 [打印本页]
作者: 科技颠覆者    时间: 5 天前
标题: [flask]自定义请求日志
前言

flask默认会在控制台输出非结构化的请求日志，如果要输出json格式的日志，并且要把请求日志写到单独的文件中，可以通过先禁用默认请求日志，然后在钩子函数中自行记录请求的方式来实现。
定义日志器

下面代码定义了两个JSON日志格式化器，JsonFormatter 的日志格式是给普通代码内利用的，会记录调用函数、调用文件等信息，AccessLogFormatter的日志格式用于记录请求日志，记录请求路径、响应状态码、响应时间等信息。
FlaskLogger通过继承logging.Logger来实现一些自定义功能，比如指定格式化器、创建日志目录等。
  1. class JsonFormatter(logging.Formatter):
  2.     def format(self, record: logging.LogRecord):
  3.         log_record = {
  4.             "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
  5.             "level": record.levelname,
  6.             "name": record.name,
  7.             "file": record.filename,
  8.             "lineno": record.lineno,
  9.             "func": record.funcName,
  10.             "message": record.getMessage(),
  11.         }
  12.         return json.dumps(log_record)
  15. class AccessLogFormatter(logging.Formatter):
  16.     def format(self, record: logging.LogRecord):
  17.         log_record = {
  18.             "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
  19.             "remote_addr": getattr(record, "remote_addr", ""),
  20.             "scheme": getattr(record, "scheme", ""),
  21.             "method": getattr(record, "method", ""),
  22.             "host": getattr(record, "host", ""),
  23.             "path": getattr(record, "path", ""),
  24.             "status": getattr(record, "status", ""),
  25.             "response_length": getattr(record, "response_length", ""),
  26.             "response_time": getattr(record, "response_time", 0),
  27.         }
  28.         return json.dumps(log_record)
  31. class FlaskLogger(logging.Logger):
  32.     """自定义日志类, 设置请求日志和普通日志两个不同的日志器
  33.    
  34.     Args:
  35.         name: str, 日志器名称, 默认为 __name__
  36.         level: int, 日志级别, 默认为 DEBUG
  37.         logfile: str, 日志文件名, 默认为 app.log
  38.         logdir: str, 日志文件目录, 默认为当前目录
  39.         access_log: bool, 是否用于记录访问日志, 默认为 False
  40.         console: bool, 是否输出到控制台, 默认为 True
  41.         json_log: bool, 是否使用json格式的日志, 默认为 True
  42.     """
  43.     def __init__(
  44.         self,
  45.         name: str = __name__,
  46.         level: int = logging.DEBUG,
  47.         logfile: str = "app.log",
  48.         logdir: str = "",
  49.         access_log: bool = False,
  50.         console: bool = True,
  51.         json_log: bool = True,
  52.     ):
  53.         super().__init__(name, level)
  54.         self.logfile = logfile
  55.         self.logdir = logdir
  56.         self.access_log = access_log
  57.         self.console = console
  58.         self.json_log = json_log
  59.         self.setup_logpath()
  60.         self.setup_handler()
  62.     def setup_logpath(self):
  63.         """设置日志文件路径, 如果创建日志器时未指定日志目录, 则使用当前目录"""
  64.         if not self.logdir:
  65.             return
  67.         p = Path(self.logdir)
  68.         if not p.exists():
  69.             try:
  70.                 p.mkdir(parents=True, exist_ok=True)
  71.             except Exception as e:
  72.                 print(f"Failed to create log directory: {e}")
  73.                 sys.exit(1)
  75.         self.logfile = p / self.logfile
  77.     def setup_handler(self):
  78.         if self.json_log:
  79.             formatter = self.set_json_formatter()
  80.         else:
  81.             formatter = self.set_plain_formatter()
  82.         handler_file = self.set_handler_file(formatter)
  83.         handler_stdout = self.set_handler_stdout(formatter)
  84.         self.addHandler(handler_file)
  85.         if self.console:
  86.             self.addHandler(handler_stdout)
  88.     def set_plain_formatter(self):
  89.         fmt = "%(asctime)s | %(levelname)s | %(name)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
  90.         datefmt = "%Y-%m-%dT%H:%M:%S%z"
  91.         return logging.Formatter(fmt, datefmt=datefmt)
  93.     def set_json_formatter(self):
  94.         """设置json格式的日志"""
  95.         if self.access_log:
  96.             return AccessLogFormatter()
  97.         return JsonFormatter()
  99.     def set_handler_stdout(self, formatter: logging.Formatter):
  100.         handler = logging.StreamHandler(sys.stdout)
  101.         handler.setFormatter(formatter)
  102.         return handler
  104.     def set_handler_file(self, formatter: logging.Formatter):
  105.         handler = TimedRotatingFileHandler(
  106.             filename=self.logfile,
  107.             when="midnight",
  108.             interval=1,
  109.             backupCount=7,
  110.             encoding="utf-8",
  111.         )
  112.         handler.setFormatter(formatter)
  113.         return handler
复制代码
实例化示例
  1. access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="access.log")
  2. logger = FlaskLogger(logdir="logs")
复制代码
钩子函数内记录请求日志

借助flask内置的钩子函数和全局对象，可以记录到每个请求的信息。
  1. from flask import g, request, Response
  2. import time
  4. @app.before_request
  5. def start_timer():
  6.     # 通过全局对象 g 来记录请求开始时间
  7.     g.start_time = time.time()
  10. @app.after_request
  11. def log_request(response: Response):
  12.     """记录每次请求的日志"""
  13.     response_length = (
  14.         response.content_length if response.content_length is not None else "-"
  15.     )
  16.     log_message = {
  17.         "remote_addr": request.remote_addr,
  18.         "method": request.method,
  19.         "scheme": request.scheme,
  20.         "host": request.host,
  21.         "path": request.path,
  22.         "status": response.status_code,
  23.         "response_length": response_length,
  24.         "response_time": round(time.time() - g.start_time, 4),
  25.     }
  26.     access_logger.info("", extra=log_message)
  27.     return response
复制代码
根本利用示例

实例化Flask对象，禁用默认日志，定义路由等
  1. from flask import Flask
  2. import traceback
  4. app = Flask(__name__)
  6. @app.errorhandler(Exception)
  7. def handle_exception(e):
  8.     """全局拦截异常"""
  9.     logger.error(f"An exception occurred, {traceback.format_exc()}", exc_info=e)
  10.     return "An error occurred", 500
  12. @app.get("/")
  13. def hello():
  14.     # 普通请求
  15.     logger.info("Hello World")
  16.     return "hello world"
  18. @app.get("/error")
  19. def raise_error():
  20.     # 模拟错误请求，观察是否全局捕获
  21.     raise Exception("Error")
  23. @app.get("/slow")
  24. def slow():
  25.     # 模拟慢请求，观察请求日志的响应时间
  26.     time.sleep(5)
  27.     return "slow"
  30. if __name__ == "__main__":
  31.     # 禁用默认的日志器
  32.     default_logger = logging.getLogger("werkzeug")
  33.     default_logger.disabled = True
  35.     app.run(host="127.0.0.1", port=5000)
复制代码
访问测试，logs目录会天生access.log和app.log文件，控制台输出示例
  1. {"@timestamp": "2025-04-26T00:26:20+0800", "level": "INFO", "name": "__main__", "file": "app.py", "lineno": 162, "func": "hello", "message": "Hello World"}
  2. {"@timestamp": "2025-04-26T00:26:20+0800", "remote_addr": "127.0.0.1", "scheme": "http", "method": "GET", "host": "127.0.0.1:5000", "path": "/", "status": 200, "response_length": 11, "response_time": 0.0003}
  3. {"@timestamp": "2025-04-26T00:26:20+0800", "level": "INFO", "name": "__main__", "file": "app.py", "lineno": 162, "func": "hello", "message": "Hello World"}
  4. {"@timestamp": "2025-04-26T00:26:20+0800", "remote_addr": "127.0.0.1", "scheme": "http", "method": "GET", "host": "127.0.0.1:5000", "path": "/", "status": 200, "response_length": 11, "response_time": 0.0003}
  5. {"@timestamp": "2025-04-26T00:29:47+0800", "remote_addr": "127.0.0.1", "scheme": "http", "method": "GET", "host": "127.0.0.1:5000", "path": "/slow", "status": 200, "response_length": 4, "response_time": 5.0002}
  6. {"@timestamp": "2025-04-26T00:31:02+0800", "level": "ERROR", "name": "__main__", "file": "app.py", "lineno": 129, "func": "handle_exception", "message": "An exception occurred, Traceback (most recent call last):\n  File "/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/lib/python3.11/site-packages/flask/app.py", line 917, in full_dispatch_request\n    rv = self.dispatch_request()\n         ^^^^^^^^^^^^^^^^^^^^^^^\n  File "/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/lib/python3.11/site-packages/flask/app.py", line 902, in dispatch_request\n    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)  # type: ignore[no-any-return]\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File "/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/demo1/app.py", line 168, in raise_error\n    raise Exception("Error")\nException: Error\n"}
  7. {"@timestamp": "2025-04-26T00:31:02+0800", "remote_addr": "127.0.0.1", "scheme": "http", "method": "GET", "host": "127.0.0.1:5000", "path": "/error", "status": 500, "response_length": 17, "response_time": 0.0011}
复制代码
完备利用示例

  1. from flask import Flask, request, g, Response import logging import sys from logging.handlers import TimedRotatingFileHandler import json from pathlib import Path import traceback import time  app = Flask(__name__)   class JsonFormatter(logging.Formatter):
  2.     def format(self, record: logging.LogRecord):
  3.         log_record = {
  4.             "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
  5.             "level": record.levelname,
  6.             "name": record.name,
  7.             "file": record.filename,
  8.             "lineno": record.lineno,
  9.             "func": record.funcName,
  10.             "message": record.getMessage(),
  11.         }
  12.         return json.dumps(log_record)
  15. class AccessLogFormatter(logging.Formatter):
  16.     def format(self, record: logging.LogRecord):
  17.         log_record = {
  18.             "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
  19.             "remote_addr": getattr(record, "remote_addr", ""),
  20.             "scheme": getattr(record, "scheme", ""),
  21.             "method": getattr(record, "method", ""),
  22.             "host": getattr(record, "host", ""),
  23.             "path": getattr(record, "path", ""),
  24.             "status": getattr(record, "status", ""),
  25.             "response_length": getattr(record, "response_length", ""),
  26.             "response_time": getattr(record, "response_time", 0),
  27.         }
  28.         return json.dumps(log_record)
  31. class FlaskLogger(logging.Logger):
  32.     """自定义日志类, 设置请求日志和普通日志两个不同的日志器
  33.    
  34.     Args:
  35.         name: str, 日志器名称, 默认为 __name__
  36.         level: int, 日志级别, 默认为 DEBUG
  37.         logfile: str, 日志文件名, 默认为 app.log
  38.         logdir: str, 日志文件目录, 默认为当前目录
  39.         access_log: bool, 是否用于记录访问日志, 默认为 False
  40.         console: bool, 是否输出到控制台, 默认为 True
  41.         json_log: bool, 是否使用json格式的日志, 默认为 True
  42.     """
  43.     def __init__(
  44.         self,
  45.         name: str = __name__,
  46.         level: int = logging.DEBUG,
  47.         logfile: str = "app.log",
  48.         logdir: str = "",
  49.         access_log: bool = False,
  50.         console: bool = True,
  51.         json_log: bool = True,
  52.     ):
  53.         super().__init__(name, level)
  54.         self.logfile = logfile
  55.         self.logdir = logdir
  56.         self.access_log = access_log
  57.         self.console = console
  58.         self.json_log = json_log
  59.         self.setup_logpath()
  60.         self.setup_handler()
  62.     def setup_logpath(self):
  63.         """设置日志文件路径, 如果创建日志器时未指定日志目录, 则使用当前目录"""
  64.         if not self.logdir:
  65.             return
  67.         p = Path(self.logdir)
  68.         if not p.exists():
  69.             try:
  70.                 p.mkdir(parents=True, exist_ok=True)
  71.             except Exception as e:
  72.                 print(f"Failed to create log directory: {e}")
  73.                 sys.exit(1)
  75.         self.logfile = p / self.logfile
  77.     def setup_handler(self):
  78.         if self.json_log:
  79.             formatter = self.set_json_formatter()
  80.         else:
  81.             formatter = self.set_plain_formatter()
  82.         handler_file = self.set_handler_file(formatter)
  83.         handler_stdout = self.set_handler_stdout(formatter)
  84.         self.addHandler(handler_file)
  85.         if self.console:
  86.             self.addHandler(handler_stdout)
  88.     def set_plain_formatter(self):
  89.         fmt = "%(asctime)s | %(levelname)s | %(name)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
  90.         datefmt = "%Y-%m-%dT%H:%M:%S%z"
  91.         return logging.Formatter(fmt, datefmt=datefmt)
  93.     def set_json_formatter(self):
  94.         """设置json格式的日志"""
  95.         if self.access_log:
  96.             return AccessLogFormatter()
  97.         return JsonFormatter()
  99.     def set_handler_stdout(self, formatter: logging.Formatter):
  100.         handler = logging.StreamHandler(sys.stdout)
  101.         handler.setFormatter(formatter)
  102.         return handler
  104.     def set_handler_file(self, formatter: logging.Formatter):
  105.         handler = TimedRotatingFileHandler(
  106.             filename=self.logfile,
  107.             when="midnight",
  108.             interval=1,
  109.             backupCount=7,
  110.             encoding="utf-8",
  111.         )
  112.         handler.setFormatter(formatter)
  113.         return handler  access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="access.log")
  114. logger = FlaskLogger(logdir="logs")  @app.errorhandler(Exception) def handle_exception(e):     """全局拦截异常"""     logger.error(f"An exception occurred, {traceback.format_exc()}", exc_info=e)     return "An error occurred", 500   @app.before_request def start_timer():     # 通过全局对象 g 来记录请求开始时间     g.start_time = time.time()   @app.after_request def log_request(response: Response):     """记录每次请求的日志"""     response_length = (         response.content_length if response.content_length is not None else "-"     )     log_message = {         "remote_addr": request.remote_addr,         "method": request.method,         "scheme": request.scheme,         "host": request.host,         "path": request.path,         "status": response.status_code,         "response_length": response_length,         "response_time": round(time.time() - g.start_time, 4),     }     access_logger.info("", extra=log_message)     return response   @app.get("/") def hello():     # 普通请求     logger.info("Hello World")     return "hello world"  @app.get("/error") def raise_error():     # 模仿错误请求，观察是否全局捕获     raise Exception("Error")  @app.get("/slow") def slow():     # 模仿慢请求，观察请求日志的响应时间     time.sleep(5)     return "slow"   if __name__ == "__main__":     # 禁用默认的日志器     default_logger = logging.getLogger("werkzeug")     default_logger.disabled = True      app.run(host="127.0.0.1", port=5000)
复制代码
参考



免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！  更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 IT评测·应用市场-qidao123.com技术社区 (https://dis.qidao123.com/) Powered by Discuz! X3.4