refactor: improving CC, Hasteald and cohesion metrics for the codebase

ninja/api_exceptions.py

@@ -0,0 +1,43 @@
+from typing import Any, Callable, Dict, List, Optional, Type, TypeVar, Union
+
+from django.http import HttpRequest, HttpResponse
+from ninja.errors import ValidationError, ValidationErrorContext
+_E = TypeVar("_E", bound=Exception)
+Exc = Union[_E, Type[_E]]
+ExcHandler = Callable[[HttpRequest, Exc[_E]], HttpResponse]
+
+class ExceptionRegistry:
+    def __init__(self):
+        self._handlers: Dict[Exc, ExcHandler] = {}
+
+    def add_handler(self, exc_class: Type[_E], handler: ExcHandler[_E]) -> None:
+        assert issubclass(exc_class, Exception)
+        self._handlers[exc_class] = handler
+
+    def lookup(self, exc: Exc[_E]) -> Optional[ExcHandler[_E]]:
+        for cls in type(exc).__mro__:
+            if cls in self._handlers:
+                return self._handlers[cls]
+        return None
+
+    def validation_error_from_contexts(
+        self, error_contexts: List[ValidationErrorContext]
+    ) -> ValidationError:
+        errors: List[Dict[str, Any]] = []
+        for context in error_contexts:
+            model = context.model
+            e = context.pydantic_validation_error
+            for i in e.errors(include_url=False):
+                i["loc"] = (
+                    model.__ninja_param_source__,
+                ) + model.__ninja_flatten_map_reverse__.get(i["loc"], i["loc"])
+                # removing pydantic hints
+                del i["input"]  # type: ignore
+                if (
+                    "ctx" in i
+                    and "error" in i["ctx"]
+                    and isinstance(i["ctx"]["error"], Exception)
+                ):
+                    i["ctx"]["error"] = str(i["ctx"]["error"])
+                errors.append(dict(i))
+        return ValidationError(errors)

ninja/api_registry.py

@@ -0,0 +1,26 @@
+import os
+from typing import List
+
+from ninja.errors import ConfigError
+from ninja.utils import is_debug_server
+
+
+class ApiRegistry:
+    _registry: List[str] = []
+    _imported_while_running_in_debug_server = is_debug_server()
+
+    @classmethod
+    def validate_namespace(cls, urls_namespace: str) -> None:
+        skip_registry = os.environ.get("NINJA_SKIP_REGISTRY", False)
+        if (
+            not skip_registry
+            and urls_namespace in cls._registry
+            and not cls.debug_server_url_reimport()
+        ):
+            msg = f"..."
+            raise ConfigError(msg.strip())
+        cls._registry.append(urls_namespace)
+
+    @classmethod
+    def debug_server_url_reimport(cls) -> bool:
+        return cls._imported_while_running_in_debug_server and not is_debug_server()

ninja/api_responses.py

@@ -0,0 +1,26 @@
+from typing import Any, Optional
+from django.http import HttpRequest, HttpResponse
+
+class ResponseFactory:
+    def __init__(self, renderer):
+        self.renderer = renderer
+
+    def create_response(
+        self,
+        request: HttpRequest,
+        data: Any,
+        *,
+        status: int,
+        temporal_response: Optional[HttpResponse] = None,
+    ) -> HttpResponse:
+        content = self.renderer.render(request, data, response_status=status)
+        if temporal_response:
+            temporal_response.content = content
+            return temporal_response
+        return HttpResponse(content, status=status, content_type=self.get_content_type())
+
+    def create_temporal_response(self) -> HttpResponse:
+        return HttpResponse("", content_type=self.get_content_type())
+
+    def get_content_type(self) -> str:
+        return f"{self.renderer.media_type}; charset={self.renderer.charset}"

ninja/api_routing.py

@@ -0,0 +1,71 @@
+from typing import (
+    Any,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
+from ninja.throttling import BaseThrottle
+from django.urls import URLPattern, URLResolver, reverse
+from ninja.constants import NOT_SET, NOT_SET_TYPE
+from ninja.router import Router
+from ninja.openapi.urls import get_openapi_urls, get_root_url
+from ninja.utils import normalize_path
+from django.utils.module_loading import import_string
+
+class RouterManager:
+    def __init__(self, api: "NinjaAPI", default_router: Optional[Router] = None):
+        self.api = api
+        self._routers: List[Tuple[str, Router]] = []
+        self.default_router = default_router or Router()
+        self.add_router("", self.default_router)
+
+    def add_router(
+        self,
+        prefix: str,
+        router: Union[Router, str],
+        *,
+        auth: Any = NOT_SET,
+        throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
+        tags: Optional[List[str]] = None,
+        parent_router: Optional[Router] = None,
+    ) -> None:
+        if isinstance(router, str):
+            router = import_string(router)
+            assert isinstance(router, Router)
+
+        if auth is not NOT_SET:
+            router.auth = auth
+
+        if throttle is not NOT_SET:
+            router.throttle = throttle
+
+        if tags is not None:
+            router.tags = tags
+
+        # Inherit API-level decorators from default router
+        # Prepend API decorators so they execute first (outer decorators)
+        router._decorators = self.default_router._decorators + router._decorators
+
+        if parent_router:
+            parent_prefix = next(
+                (path for path, r in self._routers if r is parent_router), None
+            )  # pragma: no cover
+            assert parent_prefix is not None
+            prefix = normalize_path("/".join((parent_prefix, prefix))).lstrip("/")
+
+        self._routers.extend(router.build_routers(prefix))
+        router.set_api_instance(self.api, parent_router)
+
+    def _get_urls(self) -> List[Union[URLResolver, URLPattern]]:
+        result = get_openapi_urls(self.api)
+
+        for prefix, router in self._routers:
+            result.extend(router.urls_paths(prefix))
+
+        result.append(get_root_url(self.api))
+        return result
+
+    def get_root_path(self, path_params):
+        name = f"{self.api.urls_namespace}:api-root"
+        return reverse(name, kwargs=path_params)