get_request_response_logger

Author: pycub

fastapi_loguru/middlewares/request_response.py

@@ -1,21 +1,23 @@
-from loguru import logger as loguru_logger
-from logging import Logger
+import time
 from http import HTTPStatus
 from typing import Callable, List, Optional
+
 from fastapi import FastAPI, Request, Response
+from loguru._logger import Logger
+from loguru import logger
 from starlette.middleware.base import BaseHTTPMiddleware
-import time
-import sys
+
+
+def get_request_response_logger():
+    fmt: str = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <level>{message}</level>"
+    logger.add(sink=sys.stdout, level="INFO", format=fmt)
+    return logger
 
 
 class RequestResponseLoggerMiddleware(BaseHTTPMiddleware):
-    def __init__(self, app: FastAPI, *, logger: Optional[Logger] = None, skip_routes: Optional[List] = []):
-        self._skip_routes = skip_routes
-        self.logger = logger
-        if not self.logger:
-            self.logger = loguru_logger
-            fmt = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <level>{message}</level>"
-            self.logger.add(sys.stdout, level="INFO", format=fmt)
+    def __init__(self, app: FastAPI, *, logger: Optional[Logger] = None, skip_routes: List[str] = []):
+        self._skip_routes: List[str] = skip_routes
+        self.logger: Logger = logger if logger else get_request_response_logger()
         super().__init__(app)
 
     async def dispatch(self, request: Request, call_next: Callable) -> Response:
@@ -31,12 +33,17 @@ def _should_skip_route(self, request: Request) -> bool:
 
     async def _log_it(self, *, request: Request, call_next: Callable) -> Response:
         begin = time.perf_counter()
-        response = await call_next(request)
+        try:
+            response = await call_next(request)
+        except Exception:
+            self.logger.exception(f"Request failed at {request.method} {request.url.path}")
+            raise
         end = time.perf_counter()
         self.logger.info(self._generate_log(request=request, response=response, response_time=end - begin))
         return response
 
     def _generate_log(self, *, request: Request, response: Response, response_time: float) -> str:
         status = response.status_code
         response_time *= 1000
-        return f"{request.method} {request.url.path} {status} {HTTPStatus(status).name} {int(response_time)}ms"
+        http_status_name = HTTPStatus(status).name  # pylint: disable=no-member
+        return f"{request.method} {request.url.path} {status} {http_status_name} {int(response_time)}ms"