[flask]自定义请求日志
前言
flask默认会在控制台输出非结构化的请求日志，如果要输出json格式的日志，并且要把请求日志写到单独的文件中，可以通过先禁用默认请求日志，然后在钩子函数中自行记录请求的方式来实现。
定义日志器
下面代码定义了两个JSON日志格式化器，JsonFormatter 的日志格式是给普通代码内使用的，会记录调用函数、调用文件等信息，AccessLogFormatter的日志格式用于记录请求日志，记录请求路径、响应状态码、响应时间等信息。
FlaskLogger通过继承logging.Logger来实现一些自定义功能，比如指定格式化器、创建日志目录等。
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord):
log_record = {
"@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
"level": record.levelname,
"name": record.name,
"file": record.filename,
"lineno": record.lineno,
"func": record.funcName,
"message": record.getMessage(),
}
return json.dumps(log_record)
class AccessLogFormatter(logging.Formatter):
def format(self, record: logging.LogRecord):
log_record = {
"@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
"remote_addr": getattr(record, "remote_addr", ""),
"scheme": getattr(record, "scheme", ""),
"method": getattr(record, "method", ""),
"host": getattr(record, "host", ""),
"path": getattr(record, "path", ""),
"status": getattr(record, "status", ""),
"response_length": getattr(record, "response_length", ""),
"response_time": getattr(record, "response_time", 0),
}
return json.dumps(log_record)
class FlaskLogger(logging.Logger):
"""自定义日志类, 设置请求日志和普通日志两个不同的日志器
Args:
name: str, 日志器名称, 默认为 __name__
level: int, 日志级别, 默认为 DEBUG
logfile: str, 日志文件名, 默认为 app.log
logdir: str, 日志文件目录, 默认为当前目录
access_log: bool, 是否用于记录访问日志, 默认为 False
console: bool, 是否输出到控制台, 默认为 True
json_log: bool, 是否使用json格式的日志, 默认为 True
"""
def __init__(
self,
name: str = __name__,
level: int = logging.DEBUG,
logfile: str = "app.log",
logdir: str = "",
access_log: bool = False,
console: bool = True,
json_log: bool = True,
):
super().__init__(name, level)
self.logfile = logfile
self.logdir = logdir
self.access_log = access_log
self.console = console
self.json_log = json_log
self.setup_logpath()
self.setup_handler()
def setup_logpath(self):
"""设置日志文件路径, 如果创建日志器时未指定日志目录, 则使用当前目录"""
if not self.logdir:
return
p = Path(self.logdir)
if not p.exists():
try:
p.mkdir(parents=True, exist_ok=True)
except Exception as e:
print(f"Failed to create log directory: {e}")
sys.exit(1)
self.logfile = p / self.logfile
def setup_handler(self):
if self.json_log:
formatter = self.set_json_formatter()
else:
formatter = self.set_plain_formatter()
handler_file = self.set_handler_file(formatter)
handler_stdout = self.set_handler_stdout(formatter)
self.addHandler(handler_file)
if self.console:
self.addHandler(handler_stdout)
def set_plain_formatter(self):
fmt = "%(asctime)s | %(levelname)s | %(name)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
datefmt = "%Y-%m-%dT%H:%M:%S%z"
return logging.Formatter(fmt, datefmt=datefmt)
def set_json_formatter(self):
"""设置json格式的日志"""
if self.access_log:
return AccessLogFormatter()
return JsonFormatter()
def set_handler_stdout(self, formatter: logging.Formatter):
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(formatter)
return handler
def set_handler_file(self, formatter: logging.Formatter):
handler = TimedRotatingFileHandler(
filename=self.logfile,
when="midnight",
interval=1,
backupCount=7,
encoding="utf-8",
)
handler.setFormatter(formatter)
return handler
实例化示例
access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="access.log")
logger = FlaskLogger(logdir="logs")
钩子函数内记录请求日志
借助flask内置的钩子函数和全局对象，可以记录到每个请求的信息。
from flask import g, request, Response
import time
@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
基本使用示例
实例化Flask对象，禁用默认日志，定义路由等
from flask import Flask
import traceback
app = Flask(__name__)
@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.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)
访问测试，logs目录会生成access.log和app.log文件，控制台输出示例
{"@timestamp": "2025-04-26T00:26:20+0800", "level": "INFO", "name": "__main__", "file": "app.py", "lineno": 162, "func": "hello", "message": "Hello World"}
{"@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}
{"@timestamp": "2025-04-26T00:26:20+0800", "level": "INFO", "name": "__main__", "file": "app.py", "lineno": 162, "func": "hello", "message": "Hello World"}
{"@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}
{"@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}
{"@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"}
{"@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}
完整使用示例
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):
def format(self, record: logging.LogRecord):
log_record = {
"@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
"level": record.levelname,
"name": record.name,
"file": record.filename,
"lineno": record.lineno,
"func": record.funcName,
"message": record.getMessage(),
}
return json.dumps(log_record)
class AccessLogFormatter(logging.Formatter):
def format(self, record: logging.LogRecord):
log_record = {
"@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
"remote_addr": getattr(record, "remote_addr", ""),
"scheme": getattr(record, "scheme", ""),
"method": getattr(record, "method", ""),
"host": getattr(record, "host", ""),
"path": getattr(record, "path", ""),
"status": getattr(record, "status", ""),
"response_length": getattr(record, "response_length", ""),
"response_time": getattr(record, "response_time", 0),
}
return json.dumps(log_record)
class FlaskLogger(logging.Logger):
"""自定义日志类, 设置请求日志和普通日志两个不同的日志器
Args:
name: str, 日志器名称, 默认为 __name__
level: int, 日志级别, 默认为 DEBUG
logfile: str, 日志文件名, 默认为 app.log
logdir: str, 日志文件目录, 默认为当前目录
access_log: bool, 是否用于记录访问日志, 默认为 False
console: bool, 是否输出到控制台, 默认为 True
json_log: bool, 是否使用json格式的日志, 默认为 True
"""
def __init__(
self,
name: str = __name__,
level: int = logging.DEBUG,
logfile: str = "app.log",
logdir: str = "",
access_log: bool = False,
console: bool = True,
json_log: bool = True,
):
super().__init__(name, level)
self.logfile = logfile
self.logdir = logdir
self.access_log = access_log
self.console = console
self.json_log = json_log
self.setup_logpath()
self.setup_handler()
def setup_logpath(self):
"""设置日志文件路径, 如果创建日志器时未指定日志目录, 则使用当前目录"""
if not self.logdir:
return
p = Path(self.logdir)
if not p.exists():
try:
p.mkdir(parents=True, exist_ok=True)
except Exception as e:
print(f"Failed to create log directory: {e}")
sys.exit(1)
self.logfile = p / self.logfile
def setup_handler(self):
if self.json_log:
formatter = self.set_json_formatter()
else:
formatter = self.set_plain_formatter()
handler_file = self.set_handler_file(formatter)
handler_stdout = self.set_handler_stdout(formatter)
self.addHandler(handler_file)
if self.console:
self.addHandler(handler_stdout)
def set_plain_formatter(self):
fmt = "%(asctime)s | %(levelname)s | %(name)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
datefmt = "%Y-%m-%dT%H:%M:%S%z"
return logging.Formatter(fmt, datefmt=datefmt)
def set_json_formatter(self):
"""设置json格式的日志"""
if self.access_log:
return AccessLogFormatter()
return JsonFormatter()
def set_handler_stdout(self, formatter: logging.Formatter):
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(formatter)
return handler
def set_handler_file(self, formatter: logging.Formatter):
handler = TimedRotatingFileHandler(
filename=self.logfile,
when="midnight",
interval=1,
backupCount=7,
encoding="utf-8",
)
handler.setFormatter(formatter)
return handler
access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="access.log")
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)