Merge pull request #76 from Lancetnik/main

refactor: little LoggingMiddleware changes to save few processor ticks

Author: lesnik512

microbootstrap/middlewares/fastapi.py

@@ -6,34 +6,37 @@
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 
 from microbootstrap.instruments.logging_instrument import fill_log_message
+from .utils import optimize_exclude_paths
 
 
 def build_fastapi_logging_middleware(
     exclude_endpoints: typing.Iterable[str],
 ) -> type[BaseHTTPMiddleware]:
+    endpoints_to_ignore: typing.Collection[str] = optimize_exclude_paths(exclude_endpoints)
+
     class FastAPILoggingMiddleware(BaseHTTPMiddleware):
         async def dispatch(
             self,
             request: fastapi.Request,
             call_next: RequestResponseEndpoint,
         ) -> fastapi.Response:
             request_path: typing.Final = request.url.path.removesuffix("/")
-            should_log: typing.Final = not any(
-                exclude_endpoint == request_path for exclude_endpoint in exclude_endpoints
-            )
+
+            if request_path in endpoints_to_ignore:
+                return await call_next(request)
+
             start_time: typing.Final = time.perf_counter_ns()
             try:
                 response = await call_next(request)
             except Exception:  # noqa: BLE001
                 response = fastapi.Response(status_code=500)
 
-            if should_log:
-                fill_log_message(
-                    "exception" if response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR else "info",
-                    request,
-                    response.status_code,
-                    start_time,
-                )
+            fill_log_message(
+                "exception" if response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR else "info",
+                request,
+                response.status_code,
+                start_time,
+            )
             return response
 
     return FastAPILoggingMiddleware

microbootstrap/middlewares/litestar.py

@@ -8,11 +8,14 @@
 from litestar.status_codes import HTTP_500_INTERNAL_SERVER_ERROR
 
 from microbootstrap.instruments.logging_instrument import fill_log_message
+from .utils import optimize_exclude_paths
 
 
 def build_litestar_logging_middleware(
     exclude_endpoints: typing.Iterable[str],
 ) -> type[MiddlewareProtocol]:
+    endpoints_to_ignore: typing.Collection[str] = optimize_exclude_paths(exclude_endpoints)
+
     class LitestarLoggingMiddleware(MiddlewareProtocol):
         def __init__(self, app: litestar.types.ASGIApp) -> None:
             self.app = app
@@ -24,14 +27,20 @@ async def __call__(
             send_function: litestar.types.Send,
         ) -> None:
             request: typing.Final[litestar.Request] = litestar.Request(request_scope)  # type: ignore[type-arg]
+
+            request_path = request.url.path.removesuffix("/")
+
+            if request_path in endpoints_to_ignore:
+                await self.app(request_scope, receive, send_function)
+                return
+
             start_time: typing.Final[int] = time.perf_counter_ns()
 
             async def log_message_wrapper(message: litestar.types.Message) -> None:
-                request_path = request.url.path.removesuffix("/")
-                should_log: typing.Final = not any(one_endpoint == request_path for one_endpoint in exclude_endpoints)
-                if message["type"] == "http.response.start" and should_log:
-                    log_level: str = "info" if message["status"] < HTTP_500_INTERNAL_SERVER_ERROR else "exception"
-                    fill_log_message(log_level, request, message["status"], start_time)
+                if message["type"] == "http.response.start":
+                    status = message["status"]
+                    log_level: str = "info" if status < HTTP_500_INTERNAL_SERVER_ERROR else "exception"
+                    fill_log_message(log_level, request, status, start_time)
 
                 await send_function(message)
 

microbootstrap/middlewares/utils.py

@@ -0,0 +1,15 @@
+import typing
+
+
+def optimize_exclude_paths(
+    exclude_endpoints: typing.Iterable[str],
+) -> typing.Collection[str]:
+    # `in` operator is faster for tuples than for lists
+    endpoints_to_ignore: typing.Collection[str] = tuple(exclude_endpoints)
+
+    # 10 is just an empirical value, based of measuring the performance
+    # iterating over a tuple of <10 elements is faster than hashing
+    if len(endpoints_to_ignore) >= 10:  # noqa: PLR2004
+        endpoints_to_ignore = set(endpoints_to_ignore)
+
+    return endpoints_to_ignore