[flask]自定义请求日志

a54546

前言

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, Responseimport time@app.before_requestdef start_timer():    # 通过全局对象 g 来记录请求开始时间    g.start_time = time.time()@app.after_requestdef 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 Flaskimport tracebackapp = 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, Responseimport loggingimport sysfrom logging.handlers import TimedRotatingFileHandlerimport jsonfrom pathlib import Pathimport tracebackimport timeapp = 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 handleraccess_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_requestdef start_timer():    # 通过全局对象 g 来记录请求开始时间    g.start_time = time.time()@app.after_requestdef 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