added support for openapi document computation for nested route mount

Author: eadwinCode

ellar/core/routing/mount.py

@@ -1,6 +1,5 @@
 import functools
 import typing as t
-import uuid
 
 from ellar.common import ControllerBase, ModuleRouter
 from ellar.common.constants import (
@@ -11,6 +10,7 @@
 from ellar.common.types import TReceive, TScope, TSend
 from ellar.core.router_builders.base import get_controller_builder_factory
 from ellar.reflect import reflect
+from ellar.utils import get_unique_type
 from starlette.middleware import Middleware
 from starlette.routing import BaseRoute, Match, Route, Router
 from starlette.routing import Mount as StarletteMount
@@ -38,8 +38,7 @@ def __init__(
         app.routes = RouteCollection(routes)  # type:ignore
         super().__init__(path=path, app=app, name=name, middleware=[])
         self.include_in_schema = include_in_schema
-        self._current_found_route_key = f"{uuid.uuid4().hex:4}_EllarMountRoute"
-        self._control_type = control_type
+        self._control_type = control_type or get_unique_type("EllarMountDynamicType")
 
         self.user_middleware = [] if middleware is None else list(middleware)
         self._middleware_stack: t.Optional[ASGIApp] = None
@@ -50,7 +49,7 @@ def build_middleware_stack(self) -> ASGIApp:
             app = cls(app, *args, **kwargs)
         return app
 
-    def get_control_type(self) -> t.Optional[t.Type[t.Any]]:
+    def get_control_type(self) -> t.Type[t.Any]:
         return self._control_type
 
     def add_route(
@@ -69,9 +68,7 @@ def add_route(
 
         if not isinstance(route, BaseRoute) and self.get_control_type():
             reflect.define_metadata(
-                CONTROLLER_CLASS_KEY,
-                route,
-                self.get_control_type(),  # type:ignore[arg-type]
+                CONTROLLER_CLASS_KEY, route, self.get_control_type()
             )
 
         self.routes.append(route)  # type:ignore[arg-type]
@@ -93,7 +90,7 @@ def matches(self, scope: TScope) -> t.Tuple[Match, TScope]:
                 match, child_scope = route.matches(scope_copy)
                 if match == Match.FULL:
                     _child_scope.update(child_scope)
-                    _child_scope[self._current_found_route_key] = route
+                    _child_scope[str(self.get_control_type())] = route
                     return Match.FULL, _child_scope
                 elif (
                     match == Match.PARTIAL
@@ -105,7 +102,7 @@ def matches(self, scope: TScope) -> t.Tuple[Match, TScope]:
                     partial_scope.update(child_scope)
 
             if partial:
-                partial_scope[self._current_found_route_key] = partial
+                partial_scope[str(self.get_control_type())] = partial
                 return Match.PARTIAL, partial_scope
 
         return Match.NONE, {}
@@ -114,9 +111,9 @@ async def _app_handler(self, scope: TScope, receive: TReceive, send: TSend) -> N
         request_logger.debug(
             f"Executing Matched URL Handler, path={scope['path']} - '{self.__class__.__name__}'"
         )
-        route = t.cast(t.Optional[Route], scope.get(self._current_found_route_key))
+        route = t.cast(t.Optional[Route], scope.get(str(self.get_control_type())))
         if route:
-            del scope[self._current_found_route_key]
+            del scope[str(self.get_control_type())]
             await route.handle(scope, receive, send)
         else:
             mount_router = t.cast(Router, self.app)

ellar/openapi/builder.py

@@ -1,3 +1,4 @@
+import functools
 import typing as t
 
 from ellar.common import IIdentitySchemes
@@ -37,6 +38,16 @@ class DocumentOpenAPIFactory:
     def __init__(self, document_dict: t.Dict) -> None:
         self._build = document_dict
 
+    def _update_openapi_route_models(
+        self,
+        route_models: t.List[OpenAPIRoute],
+        model: OpenAPIMountDocumentation,
+        openapi_tags: AttributeDict,
+    ) -> None:
+        route_models.append(model)
+        if openapi_tags:
+            self._build.setdefault("tags", []).append(openapi_tags)
+
     def _get_openapi_route_document_models(self, app: "App") -> t.List[OpenAPIRoute]:
         openapi_route_models: t.List = []
         reflector = app.reflector
@@ -65,6 +76,9 @@ def _get_openapi_route_document_models(self, app: "App") -> t.List[OpenAPIRoute]
                         mount=route,
                         global_guards=guards or app_guards,
                         name=openapi_tags.name,
+                        global_route_models_update=functools.partial(
+                            self._update_openapi_route_models, openapi_route_models
+                        ),
                     )
                 )
                 if openapi_tags:

ellar/openapi/route_doc_models.py

@@ -1,7 +1,7 @@
 import typing as t
 from abc import ABC, abstractmethod
 
-from ellar.common.compatible import cached_property
+from ellar.common.compatible import AttributeDict, cached_property
 from ellar.common.constants import (
     GUARDS_KEY,
     METHODS_WITH_BODY,
@@ -19,7 +19,11 @@
 from ellar.common.shortcuts import normalize_path
 from ellar.core.routing import EllarMount, RouteOperation
 from ellar.core.services.reflector import reflector
-from ellar.openapi.constants import IGNORE_CONTROLLER_TYPE, OPENAPI_OPERATION_KEY
+from ellar.openapi.constants import (
+    IGNORE_CONTROLLER_TYPE,
+    OPENAPI_OPERATION_KEY,
+    OPENAPI_TAG,
+)
 from ellar.pydantic import (
     JsonSchemaValue,
     ModelField,
@@ -55,27 +59,33 @@ class OpenAPIMountDocumentation(OpenAPIRoute):
     def __init__(
         self,
         mount: t.Union[EllarMount, Mount],
+        global_route_models_update: t.Callable[
+            ["OpenAPIMountDocumentation", t.Dict], t.Any
+        ],
         name: t.Optional[str] = None,
         global_guards: t.Optional[
             t.List[t.Union[t.Type["GuardCanActivate"], "GuardCanActivate"]]
         ] = None,
+        path_prefix: t.Optional[str] = None,
     ) -> None:
         self.tag = name
         self.mount = mount
         self.path_regex, self.path_format, self.param_convertors = compile_path(
-            self.mount.path
+            normalize_path(f"/{path_prefix}/{mount.path}")
+            if path_prefix
+            else mount.path
         )
         # if there is some convertor on ModuleMount Object, then we need to convert it to ModelField
         self.global_route_parameters: t.List[ModelField] = [
             EndpointArgsModel.get_convertor_model_field(name, convertor)
             for name, convertor in self.param_convertors.items()
         ]
         self.global_guards = global_guards or []
-
-        self._routes: t.List["OpenAPIRouteDocumentation"] = self._build_routes()
+        self._global_route_models_update = global_route_models_update
+        self.routes: t.List["OpenAPIRouteDocumentation"] = self._build_routes()
 
     def _build_routes(self) -> t.List["OpenAPIRouteDocumentation"]:
-        _routes: t.List[OpenAPIRouteDocumentation] = []
+        routes: t.List[OpenAPIRouteDocumentation] = []
 
         for route in self.mount.routes:
             if isinstance(route, RouteOperation) and route.include_in_schema:
@@ -85,20 +95,40 @@ def _build_routes(self) -> t.List["OpenAPIRouteDocumentation"]:
                 if not openapi.get("tags", False):
                     openapi.update(tags=[self.tag] if self.tag else ["default"])
 
-                _routes.append(
+                routes.append(
                     OpenAPIRouteDocumentation(
                         route=route,
                         global_route_parameters=self.global_route_parameters,
                         guards=guards or self.global_guards,
                         **openapi,
                     )
                 )
-        return _routes
+            elif isinstance(route, EllarMount):
+                openapi_tags = AttributeDict(
+                    reflector.get(OPENAPI_TAG, route.get_control_type()) or {}
+                )
+
+                if route.name:
+                    openapi_tags.setdefault("name", route.name)
+
+                guards = reflector.get(GUARDS_KEY, route.get_control_type())
+
+                self._global_route_models_update(
+                    OpenAPIMountDocumentation(
+                        mount=route,
+                        global_guards=guards or self.global_guards,
+                        name=openapi_tags.name,
+                        global_route_models_update=self._global_route_models_update,
+                        path_prefix=self.path_format,
+                    ),
+                    openapi_tags,
+                )
+        return routes
 
     @cached_property
     def _openapi_models(self) -> t.List[ModelField]:
         _models = []
-        for route in self._routes:
+        for route in self.routes:
             _models.extend(route.get_route_models())
         return _models
 
@@ -121,7 +151,7 @@ def get_openapi_path(
             if path_prefix
             else self.path_format
         )
-        for openapi_route in self._routes:
+        for openapi_route in self.routes:
             openapi_route.get_openapi_path(
                 paths=paths,
                 security_schemes=security_schemes,

ellar/utils/__init__.py

@@ -41,8 +41,8 @@ def generate_controller_operation_unique_id(
     return hash(path + _methods + _versioning + extra_string)
 
 
-def get_unique_type() -> t.Type:
-    return type(f"DynamicType{uuid.uuid4().hex[:6]}", (), {})
+def get_unique_type(prefix: str = "DynamicType") -> t.Type:
+    return type(f"{prefix}{uuid.uuid4().hex[:6]}", (), {})
 
 
 def get_name(endpoint: t.Union[t.Callable, t.Type, object]) -> str: