Location>code7788 >text

[flask]Custom request log

Popularity:556 ℃/2025-04-26 01:02:36

Preface

Flask will output unstructured request logs in the console by default. If you want to output a json format log and write the request log to a separate file, you can do so by first disabling the default request log and then recording the requests in the hook function.

Define loggers

The following code defines two JSON log formatters.JsonFormatterThe log format is for use in ordinary code, and it will record information such as calling functions, calling files, etc.AccessLogFormatterThe log format is used to record request logs, and to record request path, response status code, response time and other information.

FlaskLoggerThrough inheritanceTo implement some custom functions, such as specifying a formatter, creating a log directory, etc.

class JsonFormatter():
     def format(self, record: ):
         log_record = {
             "@timestamp": (record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
             "level": ,
             "name": ,
             "file": ,
             "lineno": ,
             "func": ,
             "message": (),
         }
         return (log_record)


 class AccessLogFormatter():
     def format(self, record: ):
         log_record = {
             "@timestamp": (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 (log_record)


 class FlaskLogger():
     """Custom log class, set two different loggers, request log and normal log
    
     Args:
         name: str, logger name, default is __name__
         level: int, log level, default to DEBUG
         logfile: str, log file name, default is
         logdir: str, log file directory, default to the current directory
         access_log: bool, whether it is used to record access logs, default is False
         console: bool, whether to output to the console, default is True
         json_log: bool, whether to use json format logs, default is True
     """
     def __init__(
         Self,
         name: str = __name__,
         level: int = ,
         logfile: str = "",
         logdir: str = "",
         access_log: bool = False,
         console: bool = True,
         json_log: bool = True,
     ):
         super().__init__(name, level)
          = logfile
          = logdir
         self.access_log = access_log
          = console
         self.json_log = json_log
         self.setup_logpath()
         self.setup_handler()

     def setup_logpath(self):
         """Set the log file path. If the log directory is not specified when creating the logger, use the current directory """
         if not:
             Return

         p = Path()
         if not ():
             try:
                 (parents=True, exist_ok=True)
             except Exception as e:
                 print(f"Failed to create log directory: {e}")
                 (1)

          = p /

     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)
         (handler_file)
         if :
             (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 (fmt, datefmt=datefmt)

     def set_json_formatter(self):
         """Set the log in json format"""
         if self.access_log:
             return AccessLogFormatter()
         return JsonFormatter()

     def set_handler_stdout(self, formatter: ):
         handler = ()
         (formatter)
         Return handler

     def set_handler_file(self, formatter: ):
         handler = TimedRotatingFileHandler(
             filename=,
             When="midnight",
             interval=1,
             backupCount=7,
             encoding="utf-8",
         )
         (formatter)
         Return handler

Instantiation example

access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="")
logger = FlaskLogger(logdir="logs")

Record request logs in hook function

With the help of flask's built-in hook functions and global objects, you can record the information of each request.

from flask import g, request, Response
 import time

 @app.before_request
 def start_timer():
     # Record the request start time through global object g
     g.start_time = ()


 @app.after_request
 def log_request(response: Response):
     """Login for each request"""
     response_length = (
         response.content_length if response.content_length is not None else "-"
     )
     log_message = {
         "remote_addr": request.remote_addr,
         "method": ,
         "scheme": ,
         "host": ,
         "path": ,
         "status": response.status_code,
         "response_length": response_length,
         "response_time": round(() - g.start_time, 4),
     }
     access_logger.info("", extra=log_message)
     Return response

Basic usage examples

Instantiate Flask objects, disable default logs, define routes, etc.

from flask import Flask
 import traceback

 app = Flask(__name__)

 @(Exception)
 def handle_exception(e):
     ""Global interception exception"""
     (f"An exception occurred, {traceback.format_exc()}", exc_info=e)
     return "An error occurred", 500

 @("/")
 def hello():
     # Normal request
     ("Hello World")
     return "hello world"

 @("/error")
 def raise_error():
     # Simulate the wrong request and observe whether it is globally captured
     raise Exception("Error")

 @("/slow")
 def slow():
     # Simulate slow requests and observe the response time of the request log
     (5)
     return "slow"


 if __name__ == "__main__":
     # Disable the default logger
     default_logger = ("werkzeug")
     default_logger.disabled = True

     (host="127.0.0.1", port=5000)

Access the test, the logs directory will be generatedandFile, console output example

{"@timestamp": "2025-04-26T00:26:20+0800", "level": "INFO", "name": "__main__", "file": "", "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": "", "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": "", "lineno": 129, "func": "handle_exception", "message": "An exception occurred, Traceback (most recent call last):\n  File \"/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/lib/python3.11/site-packages/flask/\", line 917, in full_dispatch_request\n    rv = self.dispatch_request()\n         ^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/lib/python3.11/site-packages/flask/\", line 902, in dispatch_request\n    return self.ensure_sync(self.view_functions[])(**view_args)  # type: ignore[no-any-return]\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/demo1/\", 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}

Complete usage example

from flask import Flask, request, g, Response
 import logging
 import sys
 from import TimedRotatingFileHandler
 import json
 from pathlib import Path
 import traceback
 import time

 app = Flask(__name__)


 class JsonFormatter():
     def format(self, record: ):
         log_record = {
             "@timestamp": (record, "%Y-%m-%dT%H:%M:%S%z"), # format iso 8601
             "level": ,
             "name": ,
             "file": ,
             "lineno": ,
             "func": ,
             "message": (),
         }
         return (log_record)


 class AccessLogFormatter():
     def format(self, record: ):
         log_record = {
             "@timestamp": (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 (log_record)


 class FlaskLogger():
     """Custom log class, set two different loggers, request log and normal log
    
     Args:
         name: str, logger name, default is __name__
         level: int, log level, default to DEBUG
         logfile: str, log file name, default is
         logdir: str, log file directory, default to the current directory
         access_log: bool, whether it is used to record access logs, default is False
         console: bool, whether to output to the console, default is True
         json_log: bool, whether to use json format logs, default is True
     """
     def __init__(
         Self,
         name: str = __name__,
         level: int = ,
         logfile: str = "",
         logdir: str = "",
         access_log: bool = False,
         console: bool = True,
         json_log: bool = True,
     ):
         super().__init__(name, level)
          = logfile
          = logdir
         self.access_log = access_log
          = console
         self.json_log = json_log
         self.setup_logpath()
         self.setup_handler()

     def setup_logpath(self):
         """Set the log file path. If the log directory is not specified when creating the logger, use the current directory """
         if not:
             Return

         p = Path()
         if not ():
             try:
                 (parents=True, exist_ok=True)
             except Exception as e:
                 print(f"Failed to create log directory: {e}")
                 (1)

          = p /

     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)
         (handler_file)
         if :
             (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 (fmt, datefmt=datefmt)

     def set_json_formatter(self):
         """Set the log in json format"""
         if self.access_log:
             return AccessLogFormatter()
         return JsonFormatter()

     def set_handler_stdout(self, formatter: ):
         handler = ()
         (formatter)
         Return handler

     def set_handler_file(self, formatter: ):
         handler = TimedRotatingFileHandler(
             filename=,
             When="midnight",
             interval=1,
             backupCount=7,
             encoding="utf-8",
         )
         (formatter)
         Return handler


 access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="")
 logger = FlaskLogger(logdir="logs")


 @(Exception)
 def handle_exception(e):
     ""Global interception exception"""
     (f"An exception occurred, {traceback.format_exc()}", exc_info=e)
     return "An error occurred", 500


 @app.before_request
 def start_timer():
     # Record the request start time through global object g
     g.start_time = ()


 @app.after_request
 def log_request(response: Response):
     """Login for each request"""
     response_length = (
         response.content_length if response.content_length is not None else "-"
     )
     log_message = {
         "remote_addr": request.remote_addr,
         "method": ,
         "scheme": ,
         "host": ,
         "path": ,
         "status": response.status_code,
         "response_length": response_length,
         "response_time": round(() - g.start_time, 4),
     }
     access_logger.info("", extra=log_message)
     Return response


 @("/")
 def hello():
     # Normal request
     ("Hello World")
     return "hello world"

 @("/error")
 def raise_error():
     # Simulate the wrong request and observe whether it is globally captured
     raise Exception("Error")

 @("/slow")
 def slow():
     # Simulate slow requests and observe the response time of the request log
     (5)
     return "slow"


 if __name__ == "__main__":
     # Disable the default logger
     default_logger = ("werkzeug")
     default_logger.disabled = True

     (host="127.0.0.1", port=5000)

refer to

  • Flask - logging
  • Python - logging