Implementing Async Logging in FastAPI Middleware

Middleware FastAPI Async Logging

If you are accustomed to Python’s logging module and frequently work with large datasets, you might consider implementing logging in a way that avoids blocking the CPU during I/O operations, whether writing logs to disk or sending them over the network. It’s essential to ensure that logging operations do not significantly impact the performance of your application.

The main goal of our APIs is to return data to the users. If we implement asynchronous logging in the middleware for the API logs;

• We may experience less performance degradation and achieve better scaling.
• Additionally, developers save time as the logging middleware handles the data writing.

In this article we explore how to implement robust logging and middleware functionalities to streamline your FastAPI projects. We’ll delve into creating custom loggers, handling API responses efficiently, and incorporating middleware for enhanced request processing.

1. Custom Logging Handler (API_response_handler.py): The API_response_handler module introduces a custom logging handler class designed to manage API responses efficiently. This handler facilitates seamless integration with database operations, allowing for persistent storage of log data. By extending the logging.Handler class, it enables customized handling of log records, ensuring streamlined processing and storage.

# app/API_response_handler.py

import logging
import time
from Database import Database

class APIResponseHandler(logging.Handler):
    """
    A handler class which allows the cursor to stay on
    one line for selected messages
    """
    on_same_line = False

    #record is a LogRecord object
    def emit(self, record):
        try:
            db_logger = 'insert_log.sql'
            params = {"message":record.message, "username": record.username, "method":record.method, "service":record.service,"status_code": record.status_code}
            Database().execute_dict_insert(db_logger, params)

        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self.handleError(record)


# execute_dict_insert method in Database class:
def execute_dict_insert(self, filename:str, query_dict:dict):
        try:
            self._cur.execute(open(filename, 'r').read(), query_dict)
            self._conn.commit()
            self.disconnect()
        except Exception as error:
            raise error
        return

2. Queue-Based Logging (custom_logger.py): Utilizing a queue-based logging approach, the custom_logger module optimizes log handling by decoupling log production from log consumption. This design enhances performance by minimizing blocking during logging operations, particularly in scenarios with heavy data processing. Through integration with QueueHandler and QueueListener, it offers a scalable and efficient logging solution for FastAPI applications.

# app/custom_logger.py

import logging
import API_response_handler
from queue import SimpleQueue

queue = SimpleQueue()
queue_handler = logging.handlers.QueueHandler(queue)

# Create formatter
formatter = logging.Formatter(
    fmt= "TIME (%(asctime)s) - %(levelname)s - %(message)s - CUSTOM_LOGGER"
)

#create handlers
stream_handler = API_response_handler.APIResponseHandler()
file_handler = logging.FileHandler('app.log')

#set formatter
stream_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)

log_queue_listener = logging.handlers.QueueListener(
    queue,
    stream_handler,
    file_handler,
)

#Use your own logger
api_custom_logger = logging.getLogger('APIResponseLogger')
api_custom_logger.addHandler(queue_handler)
api_custom_logger.setLevel(logging.INFO)

log_queue_listener.start()py

3. Middleware for Request Processing (middleware.py): The middleware module introduces middleware functionality to enhance request processing in FastAPI applications. By leveraging middleware components, such as BaseHTTPMiddleware, it enables the seamless integration of additional processing logic into the request-response cycle. In this case, the middleware calculates and logs essential statistics, such as response times and service invocations, enriching the application's monitoring capabilities.

# app/middleware.py

import time
from fastapi import Request
from custom_logger import api_custom_logger
import logging
from utils.Users import *
from starlette.middleware.base import BaseHTTPMiddleware

class AppMiddleware(BaseHTTPMiddleware):
    def __init__(
            self,
            app
    ):
        super().__init__(app)


    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        logging.getLogger('APIResponseLogger').info("Successfully called API", extra={
            "username": await self.get_username(request),
            "status_code": response.status_code,
            "service": str(request.url),
            "method": request.method,
            "time":process_time
        })
        return response

4. Application Setup and Lifecycle Management (main.py): The main module orchestrates the setup and lifecycle management of the FastAPI application. It integrates custom middleware components, such as the AppMiddleware, to augment request processing with additional functionalities. Additionally, it establishes a structured approach to managing the application's lifecycle, including asynchronous context management for graceful shutdown procedures.

#app/main.py

import asyncio
from custom_logger import log_queue_listener
from middleware import AppMiddleware
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    ## Stop listening in queue
    log_queue_listener.stop()

app = FastAPI(title="EsraAPI", lifespan=lifespan)

app.add_middleware(
    AppMiddleware
)

Using async logging will mainly result two benefits:

1. Improved Performance: With asynchronous logging, log messages are queued and processed in a separate thread or process, allowing the application to continue its execution without waiting for the logging operation to complete. This separation reduces the impact of logging on the overall performance of the application, especially in high-throughput or latency-sensitive scenarios.
2. Reduced Blocking: Synchronous logging can introduce blocking behavior, where the application halts its execution while waiting for the logging operation to finish. Asynchronous logging eliminates this blocking by delegating the logging task to background threads or processes, ensuring that the main application thread can continue executing without interruptions.

Conclusion: By implementing robust logging strategies, leveraging middleware functionalities, and adopting efficient request processing techniques, FastAPI developers can enhance the reliability, performance, and monitoring capabilities of their applications. Through the exploration of custom logging handlers, queue-based logging mechanisms, and middleware integration, this article provides insights into optimizing FastAPI projects for enhanced scalability and maintainability.