Add posibility to define ecludes from logging

Author: devRoemer

README.md

@@ -37,8 +37,8 @@ app.include_router(build_default_route(ingress_base_path=config.ingress_base_pat
 Extends the asgi-logger and adds a log entry for received requests.
 Furthermore the requests are filtered, so that health checks don't spam the logs.
 
-For requests to be filtered, they need to have the header "healthcheck" with one of these values:
-"livenessprobe", "readinessprobe", "startupprobe", "prtg"
+For requests to be filtered it is possible to define substrings of the path with the optional
+parameter "excluded_pathes". (it can also be parts from the middle as it is a contains search)
 
 ```python
 from sag_py_web_common.filtered_access_logger import FilteredAccessLoggerMiddleware
@@ -49,6 +49,7 @@ app.add_middleware(
     FilteredAccessLoggerMiddleware,
     format="Completed: %(R)s - %(st)s - %(L)s",
     logger=logging.getLogger("access"),
+    excluded_pathes=["pathPart/partOne", "pathPart/partTwo"], # optional
 )
 ```
 

sag_py_web_common/filtered_access_logger.py

@@ -1,14 +1,27 @@
-from typing import Dict
+import logging
+from typing import List, Union
 
-from asgi_logger.middleware import AccessInfo, AccessLogAtoms, AccessLoggerMiddleware
-from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, HTTPScope
+from asgi_logger.middleware import (AccessInfo, AccessLogAtoms,
+                                    AccessLoggerMiddleware)
+from asgiref.typing import (ASGI3Application, ASGIReceiveCallable,
+                            ASGISendCallable, HTTPScope)
 
 
 class FilteredAccessLoggerMiddleware(AccessLoggerMiddleware):
     """The lib asgi-logger wrapped to exclude prtg and health checks from being logged
     Furthermore it adds logging of the incoming requests
     """
 
+    def __init__(
+        self,
+        app: ASGI3Application,
+        format: Union[str, None],
+        logger: Union[logging.Logger, None],
+        excluded_pathes: Union[List[str], None]
+    ) -> None:
+        super().__init__(app, format, logger)
+        self.excluded_pathes = excluded_pathes or []
+
     async def __call__(self, scope: HTTPScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
         if self._should_log(scope):
             self.logger.info("Received: %s %s", scope["method"], scope["path"])
@@ -21,13 +34,13 @@ def log(self, scope: HTTPScope, info: AccessInfo) -> None:
             self.logger.info(self.format, AccessLogAtoms(scope, info), extra=extra_args)
 
     def _should_log(self, scope: HTTPScope) -> bool:
-        return scope["type"] == "http" and not self._has_health_check_header(scope)
-
-    def _has_health_check_header(self, scope: HTTPScope) -> bool:
-        header_dict: Dict[bytes, bytes] = dict(scope["headers"])
-        return b"healthcheck" in header_dict and header_dict[b"healthcheck"] in {
-            b"livenessprobe",
-            b"readinessprobe",
-            b"startupprobe",
-            b"prtg",
-        }
+        return scope["type"] == "http" \
+            and not FilteredAccessLoggerMiddleware._is_excluded_path(scope, self.excluded_pathes)
+
+    @staticmethod
+    def _is_excluded_path(scope: HTTPScope, excluded_pathes: List[str]) -> bool:
+        if not excluded_pathes:
+            return False
+
+        path: str = str(scope["path"])
+        return any(excluded in path for excluded in excluded_pathes)

tests/test_filtered_access_logger.py

@@ -0,0 +1,67 @@
+from typing import List
+
+import pytest
+
+from sag_py_web_common.filtered_access_logger import \
+    FilteredAccessLoggerMiddleware
+
+
+@pytest.mark.parametrize("path,expected", [
+    ("/api/maintain/serviceStatus", True),
+    ("/api/maintain/serviceStatusKubernetes", True),
+    ("/api/maintain/serviceStatusPrtg", True),
+    ("/maintain/serviceStatus", True),
+    ("/maintain/serviceStatusKubernetes", True),
+    ("/maintain/serviceStatusPrtg", True),
+    ("/health/serviceStatus", True),
+    ("/health/serviceStatusKubernetes", True),
+    ("/health/serviceStatusPrtg", True),
+    ("/otherEndpoint/serviceStatus", False),
+    ("/serviceStatus", False),
+    ("/my/other/endpoint", False),
+    ("/", False)
+])
+def test_is_excluded_path_with_ignore_list(path: str, expected: bool) -> None:
+    # Arrange
+    excluded_pathes = ["maintain/serviceStatus", "health/serviceStatus"]
+    scope = {"path": path}
+
+    # Act
+    actual = FilteredAccessLoggerMiddleware._is_excluded_path(scope, excluded_pathes)  # type: ignore
+
+    # Assert
+    assert actual == expected
+
+
+@pytest.mark.parametrize("path,expected", [
+    ("/serviceStatus", False),
+    ("/my/other/endpoint", False),
+    ("/", False)
+])
+def test_is_excluded_path_without_ignore_list(path: str, expected: bool) -> None:
+    # Arrange
+    excluded_pathes = None
+    scope = {"path": path}
+
+    # Act
+    actual = FilteredAccessLoggerMiddleware._is_excluded_path(scope, excluded_pathes)  # type: ignore
+
+    # Assert
+    assert actual == expected
+
+
+@pytest.mark.parametrize("path,expected", [
+    ("/serviceStatus", False),
+    ("/my/other/endpoint", False),
+    ("/", False)
+])
+def test_is_excluded_path_with_empty_ignore_list(path: str, expected: bool) -> None:
+    # Arrange
+    excluded_pathes: List[str] = []
+    scope = {"path": path}
+
+    # Act
+    actual = FilteredAccessLoggerMiddleware._is_excluded_path(scope, excluded_pathes)  # type: ignore
+
+    # Assert
+    assert actual == expected