Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/middleware/abstract.py: 36%
128 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1import abc
2import logging
3import pathlib
4import typing as t
6from starlette.types import ASGIApp, Receive, Scope, Send
8from connexion.exceptions import MissingMiddleware, ResolverError
9from connexion.http_facts import METHODS
10from connexion.operations import AbstractOperation
11from connexion.resolver import Resolver
12from connexion.spec import Specification
14logger = logging.getLogger(__name__)
16ROUTING_CONTEXT = "connexion_routing"
19class SpecMiddleware(abc.ABC):
20 """Middlewares that need the specification(s) to be registered on them should inherit from this
21 base class"""
23 @abc.abstractmethod
24 def add_api(
25 self, specification: t.Union[pathlib.Path, str, dict], **kwargs
26 ) -> t.Any:
27 """
28 Register an API represented by a single OpenAPI specification on this middleware.
29 Multiple APIs can be registered on a single middleware.
30 """
32 @abc.abstractmethod
33 async def __call__(self, scope: Scope, receive: Receive, send: Send):
34 pass
37class AbstractSpecAPI:
38 """Base API class with only minimal behavior related to the specification."""
40 def __init__(
41 self,
42 specification: t.Union[pathlib.Path, str, dict],
43 base_path: t.Optional[str] = None,
44 resolver: t.Optional[Resolver] = None,
45 arguments: t.Optional[dict] = None,
46 uri_parser_class=None,
47 *args,
48 **kwargs,
49 ):
50 self.specification = Specification.load(specification, arguments=arguments)
51 self.uri_parser_class = uri_parser_class
53 self._set_base_path(base_path)
55 self.resolver = resolver or Resolver()
57 def _set_base_path(self, base_path: t.Optional[str] = None) -> None:
58 if base_path is not None:
59 # update spec to include user-provided base_path
60 self.specification.base_path = base_path
61 self.base_path = base_path
62 else:
63 self.base_path = self.specification.base_path
66OP = t.TypeVar("OP")
67"""Typevar representing an operation"""
70class AbstractRoutingAPI(AbstractSpecAPI, t.Generic[OP]):
71 """Base API class with shared functionality related to routing."""
73 def __init__(
74 self,
75 *args,
76 pythonic_params=False,
77 resolver_error_handler: t.Optional[t.Callable] = None,
78 **kwargs,
79 ) -> None:
80 super().__init__(*args, **kwargs)
81 self.pythonic_params = pythonic_params
82 self.resolver_error_handler = resolver_error_handler
84 self.add_paths()
86 def add_paths(self, paths: t.Optional[dict] = None) -> None:
87 """
88 Adds the paths defined in the specification as operations.
89 """
90 paths = paths or self.specification.get("paths", dict())
91 for path, methods in paths.items():
92 logger.debug("Adding %s%s...", self.base_path, path)
94 for method in methods:
95 if method not in METHODS:
96 continue
97 try:
98 self.add_operation(path, method)
99 except ResolverError as err:
100 # If we have an error handler for resolver errors, add it as an operation.
101 # Otherwise treat it as any other error.
102 if self.resolver_error_handler is not None:
103 self._add_resolver_error_handler(method, path, err)
104 else:
105 self._handle_add_operation_error(path, method, err)
106 except Exception as e:
107 # All other relevant exceptions should be handled as well.
108 self._handle_add_operation_error(path, method, e)
110 def add_operation(self, path: str, method: str) -> None:
111 """
112 Adds one operation to the api.
114 This method uses the OperationID identify the module and function that will handle the operation
116 From Swagger Specification:
118 **OperationID**
120 A friendly name for the operation. The id MUST be unique among all operations described in the API.
121 Tools and libraries MAY use the operation id to uniquely identify an operation.
122 """
123 spec_operation_cls = self.specification.operation_cls
124 spec_operation = spec_operation_cls.from_spec(
125 self.specification,
126 path=path,
127 method=method,
128 resolver=self.resolver,
129 uri_parser_class=self.uri_parser_class,
130 )
131 operation = self.make_operation(spec_operation)
132 path, name = self._framework_path_and_name(spec_operation, path)
133 self._add_operation_internal(method, path, operation, name=name)
135 @abc.abstractmethod
136 def make_operation(self, operation: AbstractOperation) -> OP:
137 """Build an operation to register on the API."""
139 @staticmethod
140 def _framework_path_and_name(
141 operation: AbstractOperation, path: str
142 ) -> t.Tuple[str, str]:
143 """Prepare the framework path & name to register the operation on the API."""
145 @abc.abstractmethod
146 def _add_operation_internal(
147 self, method: str, path: str, operation: OP, name: str = None
148 ) -> None:
149 """
150 Adds the operation according to the user framework in use.
151 It will be used to register the operation on the user framework router.
152 """
154 def _add_resolver_error_handler(
155 self, method: str, path: str, err: ResolverError
156 ) -> None:
157 """
158 Adds a handler for ResolverError for the given method and path.
159 """
160 self.resolver_error_handler = t.cast(t.Callable, self.resolver_error_handler)
161 operation = self.resolver_error_handler(
162 err,
163 )
164 self._add_operation_internal(method, path, operation)
166 def _handle_add_operation_error(
167 self, path: str, method: str, exc: Exception
168 ) -> None:
169 url = f"{self.base_path}{path}"
170 error_msg = f"Failed to add operation for {method.upper()} {url}"
171 logger.error(error_msg)
172 raise exc from None
175class RoutedAPI(AbstractSpecAPI, t.Generic[OP]):
176 def __init__(
177 self,
178 specification: t.Union[pathlib.Path, str, dict],
179 *args,
180 next_app: ASGIApp,
181 **kwargs,
182 ) -> None:
183 super().__init__(specification, *args, **kwargs)
184 self.next_app = next_app
185 self.operations: t.MutableMapping[str, OP] = {}
187 def add_paths(self) -> None:
188 paths = self.specification.get("paths", {})
189 for path, methods in paths.items():
190 for method in methods:
191 if method not in METHODS:
192 continue
193 try:
194 self.add_operation(path, method)
195 except ResolverError:
196 # ResolverErrors are either raised or handled in routing middleware.
197 pass
199 def add_operation(self, path: str, method: str) -> None:
200 operation_spec_cls = self.specification.operation_cls
201 operation = operation_spec_cls.from_spec(
202 self.specification,
203 path=path,
204 method=method,
205 resolver=self.resolver,
206 uri_parser_class=self.uri_parser_class,
207 )
208 routed_operation = self.make_operation(operation)
209 self.operations[operation.operation_id] = routed_operation
211 @abc.abstractmethod
212 def make_operation(self, operation: AbstractOperation) -> OP:
213 """Create an operation of the `operation_cls` type."""
214 raise NotImplementedError
217API = t.TypeVar("API", bound="RoutedAPI")
218"""Typevar representing an API which subclasses RoutedAPI"""
221class RoutedMiddleware(SpecMiddleware, t.Generic[API]):
222 """Baseclass for middleware that wants to leverage the RoutingMiddleware to route requests to
223 its operations.
225 The RoutingMiddleware adds the operation_id to the ASGI scope. This middleware registers its
226 operations by operation_id at startup. At request time, the operation is fetched by an
227 operation_id lookup.
228 """
230 api_cls: t.Type[API]
231 """The subclass of RoutedAPI this middleware uses."""
233 def __init__(self, app: ASGIApp, **kwargs) -> None:
234 self.app = app
235 self.apis: t.Dict[str, API] = {}
237 def add_api(self, specification: t.Union[pathlib.Path, str, dict], **kwargs) -> API:
238 api = self.api_cls(specification, next_app=self.app, **kwargs)
239 self.apis[api.base_path] = api
240 return api
242 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
243 """Fetches the operation related to the request and calls it."""
244 if scope["type"] != "http":
245 await self.app(scope, receive, send)
246 return
248 try:
249 connexion_context = scope["extensions"][ROUTING_CONTEXT]
250 except KeyError:
251 raise MissingMiddleware(
252 "Could not find routing information in scope. Please make sure "
253 "you have a routing middleware registered upstream. "
254 )
255 api_base_path = connexion_context.get("api_base_path")
256 if api_base_path is not None and api_base_path in self.apis:
257 api = self.apis[api_base_path]
258 operation_id = connexion_context.get("operation_id")
259 try:
260 operation = api.operations[operation_id]
261 except KeyError as e:
262 if operation_id is None:
263 logger.debug("Skipping operation without id.")
264 await self.app(scope, receive, send)
265 return
266 else:
267 raise MissingOperation("Encountered unknown operation_id.") from e
268 else:
269 return await operation(scope, receive, send)
271 await self.app(scope, receive, send)
274class MissingOperation(Exception):
275 """Missing operation"""