Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import abc
2import logging
3import typing as t
4from collections import defaultdict
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(self, specification: Specification, **kwargs) -> t.Any:
25 """
26 Register an API represented by a single OpenAPI specification on this middleware.
27 Multiple APIs can be registered on a single middleware.
28 """
30 @abc.abstractmethod
31 async def __call__(self, scope: Scope, receive: Receive, send: Send):
32 pass
35class AbstractSpecAPI:
36 """Base API class with only minimal behavior related to the specification."""
38 def __init__(
39 self,
40 specification: Specification,
41 base_path: t.Optional[str] = None,
42 resolver: t.Optional[Resolver] = None,
43 uri_parser_class=None,
44 *args,
45 **kwargs,
46 ):
47 self.specification = specification
48 self.uri_parser_class = uri_parser_class
50 self._set_base_path(base_path)
52 self.resolver = resolver or Resolver()
54 def _set_base_path(self, base_path: t.Optional[str] = None) -> None:
55 if base_path is not None:
56 # update spec to include user-provided base_path
57 self.specification.base_path = base_path
58 self.base_path = base_path
59 else:
60 self.base_path = self.specification.base_path
63OP = t.TypeVar("OP")
64"""Typevar representing an operation"""
67class AbstractRoutingAPI(AbstractSpecAPI, t.Generic[OP]):
68 """Base API class with shared functionality related to routing."""
70 def __init__(
71 self,
72 *args,
73 pythonic_params=False,
74 resolver_error_handler: t.Optional[t.Callable] = None,
75 **kwargs,
76 ) -> None:
77 super().__init__(*args, **kwargs)
78 self.pythonic_params = pythonic_params
79 self.resolver_error_handler = resolver_error_handler
81 self.add_paths()
83 def add_paths(self, paths: t.Optional[dict] = None) -> None:
84 """
85 Adds the paths defined in the specification as operations.
86 """
87 paths = t.cast(dict, paths or self.specification.get("paths", dict()))
88 for path, methods in paths.items():
89 logger.debug("Adding %s%s...", self.base_path, path)
91 for method in methods:
92 if method not in METHODS:
93 continue
94 try:
95 self.add_operation(path, method)
96 except ResolverError as err:
97 # If we have an error handler for resolver errors, add it as an operation.
98 # Otherwise treat it as any other error.
99 if self.resolver_error_handler is not None:
100 self._add_resolver_error_handler(method, path, err)
101 else:
102 self._handle_add_operation_error(path, method, err)
103 except Exception as e:
104 # All other relevant exceptions should be handled as well.
105 self._handle_add_operation_error(path, method, e)
107 def add_operation(self, path: str, method: str) -> None:
108 """
109 Adds one operation to the api.
111 This method uses the OperationID identify the module and function that will handle the operation
113 From Swagger Specification:
115 **OperationID**
117 A friendly name for the operation. The id MUST be unique among all operations described in the API.
118 Tools and libraries MAY use the operation id to uniquely identify an operation.
119 """
120 spec_operation_cls = self.specification.operation_cls
121 spec_operation = spec_operation_cls.from_spec(
122 self.specification,
123 path=path,
124 method=method,
125 resolver=self.resolver,
126 uri_parser_class=self.uri_parser_class,
127 )
128 operation = self.make_operation(spec_operation)
129 path, name = self._framework_path_and_name(spec_operation, path)
130 self._add_operation_internal(method, path, operation, name=name)
132 @abc.abstractmethod
133 def make_operation(self, operation: AbstractOperation) -> OP:
134 """Build an operation to register on the API."""
136 @staticmethod
137 @abc.abstractmethod
138 def _framework_path_and_name(
139 operation: AbstractOperation, path: str
140 ) -> t.Tuple[str, str]:
141 """Prepare the framework path & name to register the operation on the API."""
143 @abc.abstractmethod
144 def _add_operation_internal(
145 self, method: str, path: str, operation: OP, name: t.Optional[str] = None
146 ) -> None:
147 """
148 Adds the operation according to the user framework in use.
149 It will be used to register the operation on the user framework router.
150 """
152 def _add_resolver_error_handler(
153 self, method: str, path: str, err: ResolverError
154 ) -> None:
155 """
156 Adds a handler for ResolverError for the given method and path.
157 """
158 self.resolver_error_handler = t.cast(t.Callable, self.resolver_error_handler)
159 operation = self.resolver_error_handler(
160 err,
161 )
162 self._add_operation_internal(method, path, operation)
164 def _handle_add_operation_error(
165 self, path: str, method: str, exc: Exception
166 ) -> None:
167 url = f"{self.base_path}{path}"
168 error_msg = f"Failed to add operation for {method.upper()} {url}"
169 logger.error(error_msg)
170 raise exc from None
173class RoutedAPI(AbstractSpecAPI, t.Generic[OP]):
174 def __init__(
175 self,
176 specification: Specification,
177 *args,
178 next_app: ASGIApp,
179 **kwargs,
180 ) -> None:
181 super().__init__(specification, *args, **kwargs)
182 self.next_app = next_app
183 self.operations: t.MutableMapping[t.Optional[str], OP] = {}
185 def add_paths(self) -> None:
186 paths = self.specification.get("paths", {})
187 for path, methods in paths.items():
188 for method in methods:
189 if method not in METHODS:
190 continue
191 try:
192 self.add_operation(path, method)
193 except ResolverError:
194 # ResolverErrors are either raised or handled in routing middleware.
195 pass
197 def add_operation(self, path: str, method: str) -> None:
198 operation_spec_cls = self.specification.operation_cls
199 operation = operation_spec_cls.from_spec(
200 self.specification,
201 path=path,
202 method=method,
203 resolver=self.resolver,
204 uri_parser_class=self.uri_parser_class,
205 )
206 routed_operation = self.make_operation(operation)
207 self.operations[operation.operation_id] = routed_operation
209 @abc.abstractmethod
210 def make_operation(self, operation: AbstractOperation) -> OP:
211 """Create an operation of the `operation_cls` type."""
212 raise NotImplementedError
215API = t.TypeVar("API", bound="RoutedAPI")
216"""Typevar representing an API which subclasses RoutedAPI"""
219class RoutedMiddleware(SpecMiddleware, t.Generic[API]):
220 """Baseclass for middleware that wants to leverage the RoutingMiddleware to route requests to
221 its operations.
223 The RoutingMiddleware adds the operation_id to the ASGI scope. This middleware registers its
224 operations by operation_id at startup. At request time, the operation is fetched by an
225 operation_id lookup.
226 """
228 api_cls: t.Type[API]
229 """The subclass of RoutedAPI this middleware uses."""
231 def __init__(self, app: ASGIApp) -> None:
232 self.app = app
233 self.apis: t.Dict[str, t.List[API]] = defaultdict(list)
235 def add_api(self, specification: Specification, **kwargs) -> API:
236 api = self.api_cls(specification, next_app=self.app, **kwargs)
237 self.apis[api.base_path].append(api)
238 return api
240 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
241 """Fetches the operation related to the request and calls it."""
242 if scope["type"] != "http":
243 await self.app(scope, receive, send)
244 return
246 try:
247 connexion_context = scope["extensions"][ROUTING_CONTEXT]
248 except KeyError:
249 raise MissingMiddleware(
250 "Could not find routing information in scope. Please make sure "
251 "you have a routing middleware registered upstream. "
252 )
253 api_base_path = connexion_context.get("api_base_path")
254 if api_base_path is not None and api_base_path in self.apis:
255 for api in self.apis[api_base_path]:
256 operation_id = connexion_context.get("operation_id")
257 try:
258 operation = api.operations[operation_id]
259 except KeyError:
260 if operation_id is None:
261 logger.debug("Skipping operation without id.")
262 await self.app(scope, receive, send)
263 return
264 else:
265 return await operation(scope, receive, send)
267 raise MissingOperation("Encountered unknown operation_id.")
269 await self.app(scope, receive, send)
272class MissingOperation(Exception):
273 """Missing operation"""