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 copy
2import dataclasses
3import enum
4import logging
5import pathlib
6import typing as t
7from dataclasses import dataclass, field
8from functools import partial
10from starlette.types import ASGIApp, Receive, Scope, Send
12from connexion import utils
13from connexion.handlers import ResolverErrorHandler
14from connexion.jsonifier import Jsonifier
15from connexion.lifecycle import ConnexionRequest, ConnexionResponse
16from connexion.middleware.abstract import SpecMiddleware
17from connexion.middleware.context import ContextMiddleware
18from connexion.middleware.exceptions import ExceptionMiddleware
19from connexion.middleware.lifespan import Lifespan, LifespanMiddleware
20from connexion.middleware.request_validation import RequestValidationMiddleware
21from connexion.middleware.response_validation import ResponseValidationMiddleware
22from connexion.middleware.routing import RoutingMiddleware
23from connexion.middleware.security import SecurityMiddleware
24from connexion.middleware.server_error import ServerErrorMiddleware
25from connexion.middleware.swagger_ui import SwaggerUIMiddleware
26from connexion.options import SwaggerUIOptions
27from connexion.resolver import Resolver
28from connexion.spec import Specification
29from connexion.types import MaybeAwaitable
30from connexion.uri_parsing import AbstractURIParser
31from connexion.utils import inspect_function_arguments
33logger = logging.getLogger(__name__)
36@dataclass
37class _Options:
38 """
39 Connexion provides a lot of parameters for the user to configure the app / middleware of
40 application.
42 This class provides a central place to parse these parameters a mechanism to update them.
43 Application level arguments can be provided when instantiating the application / middleware,
44 after which they can be overwritten on an API level.
46 The defaults should only be set in this class, and set to None in the signature of user facing
47 methods. This is necessary for this class to be able to differentiate between missing and
48 falsy arguments.
49 """
51 arguments: t.Optional[dict] = None
52 auth_all_paths: t.Optional[bool] = False
53 jsonifier: t.Optional[Jsonifier] = None
54 pythonic_params: t.Optional[bool] = False
55 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None
56 resolver_error: t.Optional[int] = None
57 resolver_error_handler: t.Optional[t.Callable] = field(init=False)
58 strict_validation: t.Optional[bool] = False
59 swagger_ui_options: t.Optional[SwaggerUIOptions] = None
60 uri_parser_class: t.Optional[AbstractURIParser] = None
61 validate_responses: t.Optional[bool] = False
62 validator_map: t.Optional[dict] = None
63 security_map: t.Optional[dict] = None
65 def __post_init__(self):
66 self.resolver = (
67 Resolver(self.resolver) if callable(self.resolver) else self.resolver
68 )
69 self.resolver_error_handler = self._resolver_error_handler_factory()
71 def _resolver_error_handler_factory(
72 self,
73 ) -> t.Optional[t.Callable[[], ResolverErrorHandler]]:
74 """Returns a factory to create a ResolverErrorHandler."""
75 if self.resolver_error is not None:
77 def resolver_error_handler(*args, **kwargs) -> ResolverErrorHandler:
78 return ResolverErrorHandler(self.resolver_error, *args, **kwargs)
80 return resolver_error_handler
81 return None
83 def replace(self, **changes) -> "_Options":
84 """Update mechanism to overwrite the options. None values are discarded.
86 :param changes: Arguments accepted by the __init__ method of this class.
88 :return: An new _Options object with updated arguments.
89 """
90 changes = {key: value for key, value in changes.items() if value is not None}
91 return dataclasses.replace(self, **changes)
94class MiddlewarePosition(enum.Enum):
95 """Positions to insert a middleware"""
97 BEFORE_EXCEPTION = ExceptionMiddleware
98 """Add before the :class:`ExceptionMiddleware`. This is useful if you want your changes to
99 affect the way exceptions are handled, such as a custom error handler.
101 Be mindful that security has not yet been applied at this stage.
102 Additionally, the inserted middleware is positioned before the RoutingMiddleware, so you cannot
103 leverage any routing information yet and should implement your middleware to work globally
104 instead of on an operation level.
106 Useful for middleware which should also be applied to error responses. Note that errors
107 raised here will not be handled by the exception handlers and will always result in an
108 internal server error response.
110 :meta hide-value:
111 """
112 BEFORE_SWAGGER = SwaggerUIMiddleware
113 """Add before the :class:`SwaggerUIMiddleware`. This is useful if you want your changes to
114 affect the Swagger UI, such as a path altering middleware that should also alter the paths
115 exposed by the Swagger UI
117 Be mindful that security has not yet been applied at this stage.
119 Since the inserted middleware is positioned before the RoutingMiddleware, you cannot leverage
120 any routing information yet and should implement your middleware to work globally instead of on
121 an operation level.
123 :meta hide-value:
124 """
125 BEFORE_ROUTING = RoutingMiddleware
126 """Add before the :class:`RoutingMiddleware`. This is useful if you want your changes to be
127 applied before hitting the router, such as for path altering or CORS middleware.
129 Be mindful that security has not yet been applied at this stage.
131 Since the inserted middleware is positioned before the RoutingMiddleware, you cannot leverage
132 any routing information yet and should implement your middleware to work globally instead of on
133 an operation level.
135 :meta hide-value:
136 """
137 BEFORE_SECURITY = SecurityMiddleware
138 """Add before the :class:`SecurityMiddleware`. Insert middleware here that needs to be able to
139 adapt incoming requests before security is applied.
141 Be mindful that security has not yet been applied at this stage.
143 Since the inserted middleware is positioned after the RoutingMiddleware, you can leverage
144 routing information and implement the middleware to work on an individual operation level.
146 :meta hide-value:
147 """
148 BEFORE_VALIDATION = RequestValidationMiddleware
149 """Add before the :class:`RequestValidationMiddleware`. Insert middleware here that needs to be
150 able to adapt incoming requests before they are validated.
152 Since the inserted middleware is positioned after the RoutingMiddleware, you can leverage
153 routing information and implement the middleware to work on an individual operation level.
155 :meta hide-value:
156 """
157 BEFORE_CONTEXT = ContextMiddleware
158 """Add before the :class:`ContextMiddleware`, near the end of the stack. This is the default
159 location. The inserted middleware is only followed by the ContextMiddleware, which ensures any
160 changes to the context are properly exposed to the application.
162 Since the inserted middleware is positioned after the RoutingMiddleware, you can leverage
163 routing information and implement the middleware to work on an individual operation level.
165 Since the inserted middleware is positioned after the ResponseValidationMiddleware,
166 it can intercept responses coming from the application and alter them before they are validated.
168 :meta hide-value:
169 """
172class API:
173 def __init__(self, specification, *, base_path, **kwargs) -> None:
174 self.specification = specification
175 self.base_path = base_path
176 self.kwargs = kwargs
179class ConnexionMiddleware:
180 """The main Connexion middleware, which wraps a list of specialized middlewares around the
181 provided application."""
183 default_middlewares = [
184 ServerErrorMiddleware,
185 ExceptionMiddleware,
186 SwaggerUIMiddleware,
187 RoutingMiddleware,
188 SecurityMiddleware,
189 RequestValidationMiddleware,
190 ResponseValidationMiddleware,
191 LifespanMiddleware,
192 ContextMiddleware,
193 ]
195 def __init__(
196 self,
197 app: ASGIApp,
198 *,
199 import_name: t.Optional[str] = None,
200 lifespan: t.Optional[Lifespan] = None,
201 middlewares: t.Optional[t.List[ASGIApp]] = None,
202 specification_dir: t.Union[pathlib.Path, str] = "",
203 arguments: t.Optional[dict] = None,
204 auth_all_paths: t.Optional[bool] = None,
205 jsonifier: t.Optional[Jsonifier] = None,
206 pythonic_params: t.Optional[bool] = None,
207 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None,
208 resolver_error: t.Optional[int] = None,
209 strict_validation: t.Optional[bool] = None,
210 swagger_ui_options: t.Optional[SwaggerUIOptions] = None,
211 uri_parser_class: t.Optional[AbstractURIParser] = None,
212 validate_responses: t.Optional[bool] = None,
213 validator_map: t.Optional[dict] = None,
214 security_map: t.Optional[dict] = None,
215 ):
216 """
217 :param import_name: The name of the package or module that this object belongs to. If you
218 are using a single module, __name__ is always the correct value. If you however are
219 using a package, it’s usually recommended to hardcode the name of your package there.
220 :param middlewares: The list of middlewares to wrap around the application. Defaults to
221 :obj:`middleware.main.ConnexionmMiddleware.default_middlewares`
222 :param specification_dir: The directory holding the specification(s). The provided path
223 should either be absolute or relative to the root path of the application. Defaults to
224 the root path.
225 :param arguments: Arguments to substitute the specification using Jinja.
226 :param auth_all_paths: whether to authenticate not paths not defined in the specification.
227 Defaults to False.
228 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses.
229 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an
230 underscore is appended to any shadowed built-ins. Defaults to False.
231 :param resolver: Callable that maps operationId to a function or instance of
232 :class:`resolver.Resolver`.
233 :param resolver_error: Error code to return for operations for which the operationId could
234 not be resolved. If no error code is provided, the application will fail when trying to
235 start.
236 :param strict_validation: When True, extra form or query parameters not defined in the
237 specification result in a validation error. Defaults to False.
238 :param swagger_ui_options: Instance of :class:`options.ConnexionOptions` with
239 configuration options for the swagger ui.
240 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`.
241 :param validate_responses: Whether to validate responses against the specification. This has
242 an impact on performance. Defaults to False.
243 :param validator_map: A dictionary of validators to use. Defaults to
244 :obj:`validators.VALIDATOR_MAP`.
245 :param security_map: A dictionary of security handlers to use. Defaults to
246 :obj:`security.SECURITY_HANDLERS`.
247 """
248 import_name = import_name or str(pathlib.Path.cwd())
249 self.root_path = utils.get_root_path(import_name)
251 spec_dir = pathlib.Path(specification_dir)
252 self.specification_dir = (
253 spec_dir if spec_dir.is_absolute() else self.root_path / spec_dir
254 )
256 self.app = app
257 self.lifespan = lifespan
258 self.middlewares = (
259 middlewares
260 if middlewares is not None
261 else copy.copy(self.default_middlewares)
262 )
263 self.middleware_stack: t.Optional[t.Iterable[ASGIApp]] = None
264 self.apis: t.List[API] = []
265 self.error_handlers: t.List[tuple] = []
267 self.options = _Options(
268 arguments=arguments,
269 auth_all_paths=auth_all_paths,
270 jsonifier=jsonifier,
271 pythonic_params=pythonic_params,
272 resolver=resolver,
273 resolver_error=resolver_error,
274 strict_validation=strict_validation,
275 swagger_ui_options=swagger_ui_options,
276 uri_parser_class=uri_parser_class,
277 validate_responses=validate_responses,
278 validator_map=validator_map,
279 security_map=security_map,
280 )
282 self.extra_files: t.List[str] = []
284 def add_middleware(
285 self,
286 middleware_class: t.Type[ASGIApp],
287 *,
288 position: MiddlewarePosition = MiddlewarePosition.BEFORE_CONTEXT,
289 **options: t.Any,
290 ) -> None:
291 """Add a middleware to the stack on the specified position.
293 :param middleware_class: Middleware class to add
294 :param position: Position to add the middleware, one of the MiddlewarePosition Enum
295 :param options: Options to pass to the middleware_class on initialization
296 """
297 if self.middleware_stack is not None:
298 raise RuntimeError("Cannot add middleware after an application has started")
300 for m, middleware in enumerate(self.middlewares):
301 if isinstance(middleware, partial):
302 middleware = middleware.func
304 if middleware == position.value:
305 self.middlewares.insert(
306 m, t.cast(ASGIApp, partial(middleware_class, **options))
307 )
308 break
309 else:
310 raise ValueError(
311 f"Could not insert middleware at position {position.name}. "
312 f"Please make sure you have a {position.value} in your stack."
313 )
315 def _build_middleware_stack(self) -> t.Tuple[ASGIApp, t.Iterable[ASGIApp]]:
316 """Apply all middlewares to the provided app.
318 :return: Tuple of the outer middleware wrapping the application and a list of the wrapped
319 middlewares, including the wrapped application.
320 """
321 # Include the wrapped application in the returned list.
322 app = self.app
323 apps = [app]
324 for middleware in reversed(self.middlewares):
325 arguments, _ = inspect_function_arguments(middleware)
326 if "lifespan" in arguments:
327 app = middleware(app, lifespan=self.lifespan) # type: ignore
328 else:
329 app = middleware(app) # type: ignore
330 apps.append(app)
332 # We sort the APIs by base path so that the most specific APIs are registered first.
333 # This is due to the way Starlette matches routes.
334 self.apis = utils.sort_apis_by_basepath(self.apis)
335 for app in apps:
336 if isinstance(app, SpecMiddleware):
337 for api in self.apis:
338 app.add_api(
339 api.specification,
340 base_path=api.base_path,
341 **api.kwargs,
342 )
344 if isinstance(app, ExceptionMiddleware):
345 for error_handler in self.error_handlers:
346 app.add_exception_handler(*error_handler)
348 return app, list(reversed(apps))
350 def add_api(
351 self,
352 specification: t.Union[pathlib.Path, str, dict],
353 *,
354 base_path: t.Optional[str] = None,
355 name: t.Optional[str] = None,
356 arguments: t.Optional[dict] = None,
357 auth_all_paths: t.Optional[bool] = None,
358 jsonifier: t.Optional[Jsonifier] = None,
359 pythonic_params: t.Optional[bool] = None,
360 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None,
361 resolver_error: t.Optional[int] = None,
362 strict_validation: t.Optional[bool] = None,
363 swagger_ui_options: t.Optional[SwaggerUIOptions] = None,
364 uri_parser_class: t.Optional[AbstractURIParser] = None,
365 validate_responses: t.Optional[bool] = None,
366 validator_map: t.Optional[dict] = None,
367 security_map: t.Optional[dict] = None,
368 **kwargs,
369 ) -> None:
370 """
371 Register een API represented by a single OpenAPI specification on this middleware.
372 Multiple APIs can be registered on a single middleware.
374 :param specification: OpenAPI specification. Can be provided either as dict, a path
375 to file, or a URL.
376 :param base_path: Base path to host the API. This overrides the basePath / servers in the
377 specification.
378 :param name: Name to register the API with. If no name is passed, the base_path is used
379 as name instead.
380 :param arguments: Arguments to substitute the specification using Jinja.
381 :param auth_all_paths: whether to authenticate not paths not defined in the specification.
382 Defaults to False.
383 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses.
384 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an
385 underscore is appended to any shadowed built-ins. Defaults to False.
386 :param resolver: Callable that maps operationId to a function or instance of
387 :class:`resolver.Resolver`.
388 :param resolver_error: Error code to return for operations for which the operationId could
389 not be resolved. If no error code is provided, the application will fail when trying to
390 start.
391 :param strict_validation: When True, extra form or query parameters not defined in the
392 specification result in a validation error. Defaults to False.
393 :param swagger_ui_options: A dict with configuration options for the swagger ui. See
394 :class:`options.ConnexionOptions`.
395 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`.
396 :param validate_responses: Whether to validate responses against the specification. This has
397 an impact on performance. Defaults to False.
398 :param validator_map: A dictionary of validators to use. Defaults to
399 :obj:`validators.VALIDATOR_MAP`
400 :param security_map: A dictionary of security handlers to use. Defaults to
401 :obj:`security.SECURITY_HANDLERS`
402 :param kwargs: Additional keyword arguments to pass to the `add_api` method of the managed
403 middlewares. This can be used to pass arguments to middlewares added beyond the default
404 ones.
406 :return: The Api registered on the wrapped application.
407 """
408 if self.middleware_stack is not None:
409 raise RuntimeError("Cannot add api after an application has started")
411 if isinstance(specification, str) and (
412 specification.startswith("http://") or specification.startswith("https://")
413 ):
414 pass
415 elif isinstance(specification, (pathlib.Path, str)):
416 specification = t.cast(pathlib.Path, self.specification_dir / specification)
418 # Add specification as file to watch for reloading
419 if pathlib.Path.cwd() in specification.parents:
420 self.extra_files.append(
421 str(specification.relative_to(pathlib.Path.cwd()))
422 )
424 specification = Specification.load(specification, arguments=arguments)
426 options = self.options.replace(
427 auth_all_paths=auth_all_paths,
428 jsonifier=jsonifier,
429 swagger_ui_options=swagger_ui_options,
430 pythonic_params=pythonic_params,
431 resolver=resolver,
432 resolver_error=resolver_error,
433 strict_validation=strict_validation,
434 uri_parser_class=uri_parser_class,
435 validate_responses=validate_responses,
436 validator_map=validator_map,
437 security_map=security_map,
438 )
440 api = API(
441 specification, base_path=base_path, name=name, **options.__dict__, **kwargs
442 )
443 self.apis.append(api)
445 def add_error_handler(
446 self,
447 code_or_exception: t.Union[int, t.Type[Exception]],
448 function: t.Callable[
449 [ConnexionRequest, Exception], MaybeAwaitable[ConnexionResponse]
450 ],
451 ) -> None:
452 """
453 Register a callable to handle application errors.
455 :param code_or_exception: An exception class or the status code of HTTP exceptions to
456 handle.
457 :param function: Callable that will handle exception, may be async.
458 """
459 if self.middleware_stack is not None:
460 raise RuntimeError(
461 "Cannot add error handler after an application has started"
462 )
464 error_handler = (code_or_exception, function)
465 self.error_handlers.append(error_handler)
467 def run(self, import_string: str = None, **kwargs):
468 """Run the application using uvicorn.
470 :param import_string: application as import string (eg. "main:app"). This is needed to run
471 using reload.
472 :param kwargs: kwargs to pass to `uvicorn.run`.
473 """
474 try:
475 import uvicorn
476 except ImportError:
477 raise RuntimeError(
478 "uvicorn is not installed. Please install connexion using the uvicorn extra "
479 "(connexion[uvicorn])"
480 )
482 logger.warning(
483 f"`{self.__class__.__name__}.run` is optimized for development. "
484 "For production, run using a dedicated ASGI server."
485 )
487 app: t.Union[str, ConnexionMiddleware]
488 if import_string is not None:
489 app = import_string
490 kwargs.setdefault("reload", True)
491 kwargs["reload_includes"] = self.extra_files + kwargs.get(
492 "reload_includes", []
493 )
494 else:
495 app = self
497 uvicorn.run(app, **kwargs)
499 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
500 if self.middleware_stack is None:
501 self.app, self.middleware_stack = self._build_middleware_stack()
502 # Set so starlette router throws exceptions instead of returning error responses
503 # This instance is also passed to any lifespan handler
504 scope["app"] = self
505 await self.app(scope, receive, send)