KIT-4553 Format all

Author: oschulzSam

sag_py_web_common/__init__.py

@@ -1,3 +1,3 @@
-from .default_route import build_default_route # noqa: F401
-from .filtered_access_logger import FilteredAccessLoggerMiddleware # noqa: F401
-from .json_exception_handler import handle_unknown_exception, log_exception # noqa: F401
+from .default_route import build_default_route  # noqa: F401
+from .filtered_access_logger import FilteredAccessLoggerMiddleware  # noqa: F401
+from .json_exception_handler import handle_unknown_exception, log_exception  # noqa: F401

sag_py_web_common/default_route.py

@@ -2,7 +2,9 @@
 from fastapi.responses import RedirectResponse
 
 
-def build_default_route(default_redirect_path: str = "/swagger", ingress_base_path: str = "") -> APIRouter:
+def build_default_route(
+    default_redirect_path: str = "/swagger", ingress_base_path: str = ""
+) -> APIRouter:
     """Builds an api router for the default route / It can be used to redirect to a specific path.
 
     Args:
@@ -18,6 +20,8 @@ def build_default_route(default_redirect_path: str = "/swagger", ingress_base_pa
         methods=["GET"],
         name="default",
         include_in_schema=False,
-        endpoint=lambda _: RedirectResponse(url=f"{ingress_base_path}{default_redirect_path}"),
+        endpoint=lambda _: RedirectResponse(
+            url=f"{ingress_base_path}{default_redirect_path}"
+        ),
     )
     return default_route

sag_py_web_common/filtered_access_logger.py

@@ -2,7 +2,12 @@
 from typing import List, Optional, Union
 
 from asgi_logger.middleware import AccessInfo, AccessLogAtoms, AccessLoggerMiddleware
-from asgiref.typing import ASGI3Application, ASGIReceiveCallable, ASGISendCallable, HTTPScope
+from asgiref.typing import (
+    ASGI3Application,
+    ASGIReceiveCallable,
+    ASGISendCallable,
+    HTTPScope,
+)
 
 
 class FilteredAccessLoggerMiddleware(AccessLoggerMiddleware):
@@ -33,16 +38,27 @@ async def __call__(
     def log(self, scope: HTTPScope, info: AccessInfo) -> None:
         if self._should_log(scope):
             extra_args = {"execution_time": info["end_time"] - info["start_time"]}
-            if info["response"].get("status") and str(info["response"]["status"])[0] == "2":  # type: ignore
-                self.logger.info(self.format, AccessLogAtoms(scope, info), extra=extra_args)
+            if (
+                info["response"].get("status")
+                and str(info["response"]["status"])[0] == "2"
+            ):  # type: ignore
+                self.logger.info(
+                    self.format, AccessLogAtoms(scope, info), extra=extra_args
+                )
             else:
-                self.logger.warning(self.format, AccessLogAtoms(scope, info), extra=extra_args)
+                self.logger.warning(
+                    self.format, AccessLogAtoms(scope, info), extra=extra_args
+                )
 
     def _should_log(self, scope: HTTPScope) -> bool:
         return (
             scope["type"] == "http"
-            and not FilteredAccessLoggerMiddleware._is_excluded_via_path(scope, self.excluded_paths)
-            and not FilteredAccessLoggerMiddleware._is_excluded_via_header(scope, self.exclude_header)
+            and not FilteredAccessLoggerMiddleware._is_excluded_via_path(
+                scope, self.excluded_paths
+            )
+            and not FilteredAccessLoggerMiddleware._is_excluded_via_header(
+                scope, self.exclude_header
+            )
         )
 
     @staticmethod

sag_py_web_common/json_exception_handler.py

@@ -8,7 +8,11 @@
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from starlette.responses import Response
 
-from sag_py_web_common.response_content import SimpleDetail, ValidationErrorDetail, ValidationErrorResponse
+from sag_py_web_common.response_content import (
+    SimpleDetail,
+    ValidationErrorDetail,
+    ValidationErrorResponse,
+)
 
 logger: Logger = logging.getLogger("http_error_logger")
 
@@ -20,26 +24,35 @@ async def handle_unknown_exception(_: Any, exception: Exception) -> JSONResponse
         JSONResponse: A json response that contains the field 'detail' with the exception message.
     """
     logger.error("An unknown Error!", exc_info=True, extra={"response_status": 500})
-    return JSONResponse(status_code=500, content=SimpleDetail(detail=str(exception)).model_dump())
+    return JSONResponse(
+        status_code=500, content=SimpleDetail(detail=str(exception)).model_dump()
+    )
 
 
 async def log_exception(_, exception: StarletteHTTPException) -> Response:  # type: ignore
-    logger.error("An HTTP Error! %s", exception.detail, extra={"response_status": exception.status_code})
+    logger.error(
+        "An HTTP Error! %s",
+        exception.detail,
+        extra={"response_status": exception.status_code},
+    )
 
     return await http_exception_handler(_, exception)
 
 
-async def handle_validation_exception(_: Any, exception: RequestValidationError) -> JSONResponse:
+async def handle_validation_exception(
+    _: Any, exception: RequestValidationError
+) -> JSONResponse:
     errors = [
-        ValidationErrorDetail(**{
-            "loc": [str(loc) for loc in err["loc"]],
-            "msg": err["msg"],
-            "type": err["type"]
-        })
+        ValidationErrorDetail(
+            **{
+                "loc": [str(loc) for loc in err["loc"]],
+                "msg": err["msg"],
+                "type": err["type"],
+            }
+        )
         for err in exception.errors()
     ]
     logger.error("Validation Error!", exc_info=True, extra={"response_status": 422})
     return JSONResponse(
-        status_code=422,
-        content=ValidationErrorResponse(detail=errors).model_dump()
+        status_code=422, content=ValidationErrorResponse(detail=errors).model_dump()
     )

tests/test_filtered_access_logger.py

@@ -58,7 +58,9 @@ def test_middleware_logs_successfully(
 
     middleware.log(scope, info)  # type: ignore # pylint: disable=W0212:protected-access
 
-    logs_at_correct_log_level = [record for record in caplog.records if record.levelno == log_level]
+    logs_at_correct_log_level = [
+        record for record in caplog.records if record.levelno == log_level
+    ]
 
     assert len(caplog.records) == 1
     assert len(logs_at_correct_log_level) == 1
@@ -84,7 +86,11 @@ def test_logs_can_be_ignored_via_path_and_header(
     excluded_paths = ["maintain/serviceStatus", "health/serviceStatus"]
     mock_app: ASGI3Application = lambda scope, receive, send: None  # type: ignore # noqa: E731
     middleware = FilteredAccessLoggerMiddleware(
-        app=mock_app, format=None, logger=None, excluded_paths=excluded_paths, exclude_header=exclude_header
+        app=mock_app,
+        format=None,
+        logger=None,
+        excluded_paths=excluded_paths,
+        exclude_header=exclude_header,
     )
 
     scope = {"type": "http", "path": path, "headers": ignored_headers}
@@ -122,8 +128,13 @@ def test_excluding_log_via_path_is_possible(path: str, pathIsIgnored: bool) -> N
     assert actual == pathIsIgnored
 
 
-@pytest.mark.parametrize("path,pathIsIgnored", [("/serviceStatus", False), ("/my/other/endpoint", False), ("/", False)])
-def test_not_configured_excluded_paths_does_not_cause_any_logs_to_be_excluded(path: str, pathIsIgnored: bool) -> None:
+@pytest.mark.parametrize(
+    "path,pathIsIgnored",
+    [("/serviceStatus", False), ("/my/other/endpoint", False), ("/", False)],
+)
+def test_not_configured_excluded_paths_does_not_cause_any_logs_to_be_excluded(
+    path: str, pathIsIgnored: bool
+) -> None:
     excluded_paths = None
     scope = {"path": path}
 
@@ -133,8 +144,13 @@ def test_not_configured_excluded_paths_does_not_cause_any_logs_to_be_excluded(pa
     assert actual == pathIsIgnored
 
 
-@pytest.mark.parametrize("path,pathIsIgnored", [("/serviceStatus", False), ("/my/other/endpoint", False), ("/", False)])
-def test_empty_excludes_paths_does_not_cause_any_logs_to_be_excluded(path: str, pathIsIgnored: bool) -> None:
+@pytest.mark.parametrize(
+    "path,pathIsIgnored",
+    [("/serviceStatus", False), ("/my/other/endpoint", False), ("/", False)],
+)
+def test_empty_excludes_paths_does_not_cause_any_logs_to_be_excluded(
+    path: str, pathIsIgnored: bool
+) -> None:
     excluded_paths: List[str] = []
     scope = {"path": path}
 
@@ -154,26 +170,34 @@ def test_empty_excludes_paths_does_not_cause_any_logs_to_be_excluded(path: str,
         ([], False),
     ],
 )
-def test_excluding_logs_via_header_is_possible(headers: list[tuple[bytes, bytes]], path_is_ignored: bool) -> None:
+def test_excluding_logs_via_header_is_possible(
+    headers: list[tuple[bytes, bytes]], path_is_ignored: bool
+) -> None:
     exclude_header = "exclude-logging"
     scope = {
         "headers": headers,
     }
 
     # pylint: disable=W0212:protected-access
-    actual = FilteredAccessLoggerMiddleware._is_excluded_via_header(scope, exclude_header)  # type: ignore
+    actual = FilteredAccessLoggerMiddleware._is_excluded_via_header(
+        scope, exclude_header
+    )  # type: ignore
 
     assert actual == path_is_ignored
 
 
-@pytest.mark.parametrize("headers,path_is_ignored", [([(b"Exclude-Logging", b"")], False), ([], False)])
+@pytest.mark.parametrize(
+    "headers,path_is_ignored", [([(b"Exclude-Logging", b"")], False), ([], False)]
+)
 def test_not_configured_exclude_header_does_not_cause_any_logs_to_be_excluded(
     headers: list[tuple[bytes, bytes]], path_is_ignored: bool
 ) -> None:
     exclude_header = ""
     scope = {"headers": headers}
 
     # pylint: disable=W0212:protected-access
-    actual = FilteredAccessLoggerMiddleware._is_excluded_via_header(scope, exclude_header)  # type: ignore
+    actual = FilteredAccessLoggerMiddleware._is_excluded_via_header(
+        scope, exclude_header
+    )  # type: ignore
 
     assert actual == path_is_ignored

tests/test_json_exception_handler.py

@@ -14,7 +14,9 @@
 @pytest.mark.asyncio
 async def test_handle_unknown_exception() -> None:
     # Act
-    result: JSONResponse = await handle_unknown_exception("", Exception("error message"))
+    result: JSONResponse = await handle_unknown_exception(
+        "", Exception("error message")
+    )
 
     # Assert
     assert result.status_code == 500
@@ -24,7 +26,15 @@ async def test_handle_unknown_exception() -> None:
 @pytest.mark.asyncio
 async def test_validation_exception_handler() -> None:
     # Arrange
-    exc = RequestValidationError([{"loc": ("body", "title"), "msg": "field required", "type": "value_error.missing"}])
+    exc = RequestValidationError(
+        [
+            {
+                "loc": ("body", "title"),
+                "msg": "field required",
+                "type": "value_error.missing",
+            }
+        ]
+    )
     request = Mock()
 
     # Act