less code

Author: khvn26

src/common/gunicorn/logging.py

@@ -14,7 +14,6 @@
 from common.core.logging import JsonFormatter
 from common.gunicorn import metrics
 from common.gunicorn.constants import WSGI_DJANGO_ROUTE_ENVIRON_KEY
-from common.gunicorn.utils import get_status_from_wsgi_response_status
 from common.prometheus.utils import with_labels
 
 
@@ -57,7 +56,7 @@ def access(
             # The Django route is set by `PrometheusGunicornLoggerMiddleware`.
             "path": environ.get(WSGI_DJANGO_ROUTE_ENVIRON_KEY),
             "method": environ.get("REQUEST_METHOD"),
-            "response_status": get_status_from_wsgi_response_status(resp.status),
+            "response_status": resp.status_code,
         }
         with_labels(metrics.http_request_duration_seconds, **labels).observe(
             duration_seconds

src/common/gunicorn/utils.py

@@ -1,19 +1,14 @@
 import argparse
-from functools import lru_cache
 from typing import Any
 
 from django.core.handlers.wsgi import WSGIHandler
 from django.core.wsgi import get_wsgi_application
-from django.urls import Resolver404, resolve
 from environs import Env
 from gunicorn.app.wsgiapp import (  # type: ignore[import-untyped]
     WSGIApplication as GunicornWSGIApplication,
 )
 from gunicorn.config import Config  # type: ignore[import-untyped]
 
-from common.prometheus.constants import UNKNOWN_LABEL_VALUE
-from common.prometheus.types import LabelValue
-
 env = Env()
 
 DEFAULT_ACCESS_LOG_FORMAT = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %({origin}i)s %({access-control-allow-origin}o)s'
@@ -54,19 +49,6 @@ def load_wsgiapp(self) -> WSGIHandler:
         return get_wsgi_application()
 
 
-@lru_cache(maxsize=64)
-def get_status_from_wsgi_response_status(
-    status: str | bytes | int | None,
-) -> str | None:
-    if isinstance(status, int):
-        return str(status)
-    if isinstance(status, bytes):
-        status = status.decode("utf-8")
-    if isinstance(status, str):
-        status = status.split(None, 1)[0]
-    return status
-
-
 def add_arguments(parser: argparse._ArgumentGroup) -> None:
     _config = Config()
     keys = sorted(_config.settings, key=_config.settings.__getitem__)

src/common/prometheus/types.py

@@ -1,4 +1,3 @@
-from typing import Literal, TypeAlias
+from typing import Literal
 
 UnknownLabelValue = Literal["unknown"]
-LabelValue: TypeAlias = str | UnknownLabelValue

src/common/prometheus/utils.py

@@ -5,7 +5,6 @@
 from prometheus_client.multiprocess import MultiProcessCollector
 
 from common.prometheus.constants import UNKNOWN_LABEL_VALUE
-from common.prometheus.types import LabelValue
 
 T = typing.TypeVar("T", bound=MetricWrapperBase)
 
@@ -18,7 +17,7 @@ def get_registry() -> prometheus_client.CollectorRegistry:
 
 def with_labels(
     metric: T,
-    **labels: LabelValue | None,
+    **labels: typing.Any,
 ) -> T:
     """
     Add labels to a given metric. If a label value is `None` or is not provided,

tests/unit/common/gunicorn/test_utils.py

@@ -8,31 +8,10 @@
 
 from common.gunicorn.utils import (
     DjangoWSGIApplication,
-    get_status_from_wsgi_response_status,
     run_server,
 )
 
 
-@pytest.mark.parametrize(
-    "status,expected",
-    (
-        [b" 200 ", "200"],
-        [200, "200"],
-        ["200", "200"],
-        [None, "unknown"],
-    ),
-)
-def test_get_status_from_wsgi_response_status__returns_expected(
-    status: str | bytes | int | None,
-    expected: str,
-) -> None:
-    # When
-    result = get_status_from_wsgi_response_status(status)
-
-    # Then
-    assert result == expected
-
-
 def test_django_wsgi_application__defaults__expected_config(
     mocker: pytest_mock.MockerFixture,
 ) -> None: