Extract boilerplate code into Routed base classes

connexion/middleware/__init__.py

@@ -1,4 +1,4 @@
-from .abstract import AppMiddleware  # NOQA
+from .abstract import AppMiddleware, RoutedMiddleware  # NOQA
 from .main import ConnexionMiddleware  # NOQA
 from .routing import RoutingMiddleware  # NOQA
 from .swagger_ui import SwaggerUIMiddleware  # NOQA

connexion/middleware/abstract.py

@@ -1,7 +1,21 @@
 import abc
+import logging
 import pathlib
 import typing as t
 
+import typing_extensions as te
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from connexion.apis.abstract import AbstractSpecAPI
+from connexion.exceptions import MissingMiddleware
+from connexion.http_facts import METHODS
+from connexion.operations import AbstractOperation
+from connexion.resolver import ResolverError
+
+logger = logging.getLogger("connexion.middleware.abstract")
+
+ROUTING_CONTEXT = "connexion_routing"
+
 
 class AppMiddleware(abc.ABC):
     """Middlewares that need the APIs to be registered on them should inherit from this base
@@ -12,3 +26,117 @@ def add_api(
         self, specification: t.Union[pathlib.Path, str, dict], **kwargs
     ) -> None:
         pass
+
+
+class RoutedOperation(te.Protocol):
+    def __init__(self, next_app: ASGIApp, **kwargs) -> None:
+        ...
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        ...
+
+
+OP = t.TypeVar("OP", bound=RoutedOperation)
+
+
+class RoutedAPI(AbstractSpecAPI, t.Generic[OP]):
+
+    operation_cls: t.Type[OP]
+    """The operation this middleware uses, which should implement the RoutingOperation protocol."""
+
+    def __init__(
+        self,
+        specification: t.Union[pathlib.Path, str, dict],
+        *args,
+        next_app: ASGIApp,
+        **kwargs,
+    ) -> None:
+        super().__init__(specification, *args, **kwargs)
+        self.next_app = next_app
+        self.operations: t.MutableMapping[str, OP] = {}
+
+    def add_paths(self) -> None:
+        paths = self.specification.get("paths", {})
+        for path, methods in paths.items():
+            for method in methods:
+                if method not in METHODS:
+                    continue
+                try:
+                    self.add_operation(path, method)
+                except ResolverError:
+                    # ResolverErrors are either raised or handled in routing middleware.
+                    pass
+
+    def add_operation(self, path: str, method: str) -> None:
+        operation_spec_cls = self.specification.operation_cls
+        operation = operation_spec_cls.from_spec(
+            self.specification, self, path, method, self.resolver
+        )
+        routed_operation = self.make_operation(operation)
+        self.operations[operation.operation_id] = routed_operation
+
+    @abc.abstractmethod
+    def make_operation(self, operation: AbstractOperation) -> OP:
+        """Create an operation of the `operation_cls` type."""
+        raise NotImplementedError
+
+
+API = t.TypeVar("API", bound="RoutedAPI")
+
+
+class RoutedMiddleware(AppMiddleware, t.Generic[API]):
+    """Baseclass for middleware that wants to leverage the RoutingMiddleware to route requests to
+    its operations.
+
+    The RoutingMiddleware adds the operation_id to the ASGI scope. This middleware registers its
+    operations by operation_id at startup. At request time, the operation is fetched by an
+    operation_id lookup.
+    """
+
+    api_cls: t.Type[API]
+    """The subclass of RoutedAPI this middleware uses."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+        self.apis: t.Dict[str, API] = {}
+
+    def add_api(
+        self, specification: t.Union[pathlib.Path, str, dict], **kwargs
+    ) -> None:
+        api = self.api_cls(specification, next_app=self.app, **kwargs)
+        self.apis[api.base_path] = api
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """Fetches the operation related to the request and calls it."""
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        try:
+            connexion_context = scope["extensions"][ROUTING_CONTEXT]
+        except KeyError:
+            raise MissingMiddleware(
+                "Could not find routing information in scope. Please make sure "
+                "you have a routing middleware registered upstream. "
+            )
+        api_base_path = connexion_context.get("api_base_path")
+        if api_base_path:
+            api = self.apis[api_base_path]
+            operation_id = connexion_context.get("operation_id")
+            try:
+                operation = api.operations[operation_id]
+            except KeyError as e:
+                if operation_id is None:
+                    logger.debug("Skipping validation check for operation without id.")
+                    await self.app(scope, receive, send)
+                    return
+                else:
+                    raise MissingOperation("Encountered unknown operation_id.") from e
+            else:
+                return await operation(scope, receive, send)
+
+        await self.app(scope, receive, send)
+
+
+class MissingOperation(Exception):
+    """Missing operation"""

connexion/middleware/exceptions.py

@@ -39,7 +39,7 @@ def problem_handler(self, _, exception: ProblemException):
 
     def http_exception(self, request: Request, exc: HTTPException) -> Response:
         try:
-            headers = exc.headers
+            headers = exc.headers  # type: ignore
         except AttributeError:
             # Starlette < 0.19
             headers = {}

connexion/middleware/routing.py

@@ -6,61 +6,36 @@
 from starlette.types import ASGIApp, Receive, Scope, Send
 
 from connexion.apis import AbstractRoutingAPI
-from connexion.middleware import AppMiddleware
+from connexion.middleware.abstract import ROUTING_CONTEXT, AppMiddleware
 from connexion.operations import AbstractOperation
 from connexion.resolver import Resolver
 
-ROUTING_CONTEXT = "connexion_routing"
-
-
 _scope: ContextVar[dict] = ContextVar("SCOPE")
 
 
-class RoutingMiddleware(AppMiddleware):
-    def __init__(self, app: ASGIApp) -> None:
-        """Middleware that resolves the Operation for an incoming request and attaches it to the
-        scope.
-
-        :param app: app to wrap in middleware.
-        """
-        self.app = app
-        # Pass unknown routes to next app
-        self.router = Router(default=RoutingOperation(None, self.app))
-
-    def add_api(
-        self,
-        specification: t.Union[pathlib.Path, str, dict],
-        base_path: t.Optional[str] = None,
-        arguments: t.Optional[dict] = None,
-        **kwargs
-    ) -> None:
-        """Add an API to the router based on a OpenAPI spec.
+class RoutingOperation:
+    def __init__(self, operation_id: t.Optional[str], next_app: ASGIApp) -> None:
+        self.operation_id = operation_id
+        self.next_app = next_app
 
-        :param specification: OpenAPI spec as dict or path to file.
-        :param base_path: Base path where to add this API.
-        :param arguments: Jinja arguments to replace in the spec.
-        """
-        api = RoutingAPI(
-            specification,
-            base_path=base_path,
-            arguments=arguments,
-            next_app=self.app,
-            **kwargs
-        )
-        self.router.mount(api.base_path, app=api.router)
+    @classmethod
+    def from_operation(cls, operation: AbstractOperation, next_app: ASGIApp):
+        return cls(operation.operation_id, next_app)
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        """Route request to matching operation, and attach it to the scope before calling the
-        next app."""
-        if scope["type"] != "http":
-            await self.app(scope, receive, send)
-            return
+        """Attach operation to scope and pass it to the next app"""
+        original_scope = _scope.get()
 
-        _scope.set(scope.copy())  # type: ignore
+        api_base_path = scope.get("root_path", "")[
+            len(original_scope.get("root_path", "")) :
+        ]
 
-        # Needs to be set so starlette router throws exceptions instead of returning error responses
-        scope["app"] = self
-        await self.router(scope, receive, send)
+        extensions = original_scope.setdefault("extensions", {})
+        connexion_routing = extensions.setdefault(ROUTING_CONTEXT, {})
+        connexion_routing.update(
+            {"api_base_path": api_base_path, "operation_id": self.operation_id}
+        )
+        await self.next_app(original_scope, receive, send)
 
 
 class RoutingAPI(AbstractRoutingAPI):
@@ -105,26 +80,48 @@ def _add_operation_internal(
         self.router.add_route(path, operation, methods=[method])
 
 
-class RoutingOperation:
-    def __init__(self, operation_id: t.Optional[str], next_app: ASGIApp) -> None:
-        self.operation_id = operation_id
-        self.next_app = next_app
+class RoutingMiddleware(AppMiddleware):
+    def __init__(self, app: ASGIApp) -> None:
+        """Middleware that resolves the Operation for an incoming request and attaches it to the
+        scope.
 
-    @classmethod
-    def from_operation(cls, operation: AbstractOperation, next_app: ASGIApp):
-        return cls(operation.operation_id, next_app)
+        :param app: app to wrap in middleware.
+        """
+        self.app = app
+        # Pass unknown routes to next app
+        self.router = Router(default=RoutingOperation(None, self.app))
+
+    def add_api(
+        self,
+        specification: t.Union[pathlib.Path, str, dict],
+        base_path: t.Optional[str] = None,
+        arguments: t.Optional[dict] = None,
+        **kwargs
+    ) -> None:
+        """Add an API to the router based on a OpenAPI spec.
+
+        :param specification: OpenAPI spec as dict or path to file.
+        :param base_path: Base path where to add this API.
+        :param arguments: Jinja arguments to replace in the spec.
+        """
+        api = RoutingAPI(
+            specification,
+            base_path=base_path,
+            arguments=arguments,
+            next_app=self.app,
+            **kwargs
+        )
+        self.router.mount(api.base_path, app=api.router)
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        """Attach operation to scope and pass it to the next app"""
-        original_scope = _scope.get()
+        """Route request to matching operation, and attach it to the scope before calling the
+        next app."""
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
 
-        api_base_path = scope.get("root_path", "")[
-            len(original_scope.get("root_path", "")) :
-        ]
+        _scope.set(scope.copy())  # type: ignore
 
-        extensions = original_scope.setdefault("extensions", {})
-        connexion_routing = extensions.setdefault(ROUTING_CONTEXT, {})
-        connexion_routing.update(
-            {"api_base_path": api_base_path, "operation_id": self.operation_id}
-        )
-        await self.next_app(original_scope, receive, send)
+        # Needs to be set so starlette router throws exceptions instead of returning error responses
+        scope["app"] = self
+        await self.router(scope, receive, send)