[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)

复制代码

参考

• Flask - logging
  
• Python - logging
  

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