Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/middleware/main.py: 48%
98 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1import dataclasses
2import logging
3import pathlib
4import typing as t
5from dataclasses import dataclass, field
7from starlette.types import ASGIApp, Receive, Scope, Send
9from connexion import utils
10from connexion.handlers import ResolverErrorHandler
11from connexion.jsonifier import Jsonifier
12from connexion.middleware.abstract import SpecMiddleware
13from connexion.middleware.context import ContextMiddleware
14from connexion.middleware.exceptions import ExceptionMiddleware
15from connexion.middleware.lifespan import Lifespan, LifespanMiddleware
16from connexion.middleware.request_validation import RequestValidationMiddleware
17from connexion.middleware.response_validation import ResponseValidationMiddleware
18from connexion.middleware.routing import RoutingMiddleware
19from connexion.middleware.security import SecurityMiddleware
20from connexion.middleware.swagger_ui import SwaggerUIMiddleware
21from connexion.resolver import Resolver
22from connexion.uri_parsing import AbstractURIParser
24logger = logging.getLogger(__name__)
27@dataclass
28class _Options:
29 """
30 Connexion provides a lot of parameters for the user to configure the app / middleware of
31 application.
33 This class provides a central place to parse these parameters a mechanism to update them.
34 Application level arguments can be provided when instantiating the application / middleware,
35 after which they can be overwritten on an API level.
37 The defaults should only be set in this class, and set to None in the signature of user facing
38 methods. This is necessary for this class to be able to differentiate between missing and
39 falsy arguments.
40 """
42 arguments: t.Optional[dict] = None
43 auth_all_paths: t.Optional[bool] = False
44 jsonifier: t.Optional[Jsonifier] = None
45 pythonic_params: t.Optional[bool] = False
46 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None
47 resolver_error: t.Optional[int] = None
48 resolver_error_handler: t.Optional[t.Callable] = field(init=False)
49 strict_validation: t.Optional[bool] = False
50 swagger_ui_options: t.Optional[dict] = None
51 uri_parser_class: t.Optional[AbstractURIParser] = None
52 validate_responses: t.Optional[bool] = False
53 validator_map: t.Optional[dict] = None
55 def __post_init__(self):
56 self.resolver = (
57 Resolver(self.resolver) if callable(self.resolver) else self.resolver
58 )
59 self.resolver_error_handler = self._resolver_error_handler_factory()
61 def _resolver_error_handler_factory(
62 self,
63 ) -> t.Optional[t.Callable[[], ResolverErrorHandler]]:
64 """Returns a factory to create a ResolverErrorHandler."""
65 if self.resolver_error is not None:
67 def resolver_error_handler(*args, **kwargs) -> ResolverErrorHandler:
68 return ResolverErrorHandler(self.resolver_error, *args, **kwargs)
70 return resolver_error_handler
71 return None
73 def replace(self, **changes) -> "_Options":
74 """Update mechanism to overwrite the options. None values are discarded.
76 :param changes: Arguments accepted by the __init__ method of this class.
78 :return: An new _Options object with updated arguments.
79 """
80 changes = {key: value for key, value in changes.items() if value is not None}
81 return dataclasses.replace(self, **changes)
84class ConnexionMiddleware:
85 """The main Connexion middleware, which wraps a list of specialized middlewares around the
86 provided application."""
88 default_middlewares = [
89 ExceptionMiddleware,
90 SwaggerUIMiddleware,
91 RoutingMiddleware,
92 SecurityMiddleware,
93 RequestValidationMiddleware,
94 ResponseValidationMiddleware,
95 ContextMiddleware,
96 LifespanMiddleware,
97 ]
99 def __init__(
100 self,
101 app: ASGIApp,
102 *,
103 import_name: t.Optional[str] = None,
104 lifespan: t.Optional[Lifespan] = None,
105 middlewares: t.Optional[list] = None,
106 specification_dir: t.Union[pathlib.Path, str] = "",
107 arguments: t.Optional[dict] = None,
108 auth_all_paths: t.Optional[bool] = None,
109 jsonifier: t.Optional[Jsonifier] = None,
110 pythonic_params: t.Optional[bool] = None,
111 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None,
112 resolver_error: t.Optional[int] = None,
113 strict_validation: t.Optional[bool] = None,
114 swagger_ui_options: t.Optional[dict] = None,
115 uri_parser_class: t.Optional[AbstractURIParser] = None,
116 validate_responses: t.Optional[bool] = None,
117 validator_map: t.Optional[dict] = None,
118 ):
119 """
120 :param import_name: The name of the package or module that this object belongs to. If you
121 are using a single module, __name__ is always the correct value. If you however are
122 using a package, it’s usually recommended to hardcode the name of your package there.
123 :param middlewares: The list of middlewares to wrap around the application. Defaults to
124 :obj:`middleware.main.ConnexionmMiddleware.default_middlewares`
125 :param specification_dir: The directory holding the specification(s). The provided path
126 should either be absolute or relative to the root path of the application. Defaults to
127 the root path.
128 :param arguments: Arguments to substitute the specification using Jinja.
129 :param auth_all_paths: whether to authenticate not paths not defined in the specification.
130 Defaults to False.
131 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses.
132 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an
133 underscore is appended to any shadowed built-ins. Defaults to False.
134 :param resolver: Callable that maps operationId to a function or instance of
135 :class:`resolver.Resolver`.
136 :param resolver_error: Error code to return for operations for which the operationId could
137 not be resolved. If no error code is provided, the application will fail when trying to
138 start.
139 :param strict_validation: When True, extra form or query parameters not defined in the
140 specification result in a validation error. Defaults to False.
141 :param swagger_ui_options: A :class:`options.ConnexionOptions` instance with configuration
142 options for the swagger ui.
143 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`.
144 :param validate_responses: Whether to validate responses against the specification. This has
145 an impact on performance. Defaults to False.
146 :param validator_map: A dictionary of validators to use. Defaults to
147 :obj:`validators.VALIDATOR_MAP`.
148 """
149 import_name = import_name or str(pathlib.Path.cwd())
150 self.root_path = utils.get_root_path(import_name)
152 self.specification_dir = self.ensure_absolute(specification_dir)
154 if middlewares is None:
155 middlewares = self.default_middlewares
156 self.app, self.apps = self._apply_middlewares(
157 app, middlewares, lifespan=lifespan
158 )
160 self.options = _Options(
161 arguments=arguments,
162 auth_all_paths=auth_all_paths,
163 jsonifier=jsonifier,
164 pythonic_params=pythonic_params,
165 resolver=resolver,
166 resolver_error=resolver_error,
167 strict_validation=strict_validation,
168 swagger_ui_options=swagger_ui_options,
169 uri_parser_class=uri_parser_class,
170 validate_responses=validate_responses,
171 validator_map=validator_map,
172 )
174 self.extra_files: t.List[str] = []
176 def ensure_absolute(self, path: t.Union[str, pathlib.Path]) -> pathlib.Path:
177 """Ensure that a path is absolute. If the path is not absolute, it is assumed to relative
178 to the application root path and made absolute."""
179 path = pathlib.Path(path)
180 if path.is_absolute():
181 return path
182 else:
183 return self.root_path / path
185 def _apply_middlewares(
186 self, app: ASGIApp, middlewares: t.List[t.Type[ASGIApp]], **kwargs
187 ) -> t.Tuple[ASGIApp, t.Iterable[ASGIApp]]:
188 """Apply all middlewares to the provided app.
190 :param app: App to wrap in middlewares.
191 :param middlewares: List of middlewares to wrap around app. The list should be ordered
192 from outer to inner middleware.
194 :return: Tuple of the outer middleware wrapping the application and a list of the wrapped
195 middlewares, including the wrapped application.
196 """
197 # Include the wrapped application in the returned list.
198 apps = [app]
199 for middleware in reversed(middlewares):
200 app = middleware(app, **kwargs) # type: ignore
201 apps.append(app)
202 return app, list(reversed(apps))
204 def add_api(
205 self,
206 specification: t.Union[pathlib.Path, str, dict],
207 *,
208 base_path: t.Optional[str] = None,
209 arguments: t.Optional[dict] = None,
210 auth_all_paths: t.Optional[bool] = None,
211 jsonifier: t.Optional[Jsonifier] = None,
212 pythonic_params: t.Optional[bool] = None,
213 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None,
214 resolver_error: t.Optional[int] = None,
215 strict_validation: t.Optional[bool] = None,
216 swagger_ui_options: t.Optional[dict] = None,
217 uri_parser_class: t.Optional[AbstractURIParser] = None,
218 validate_responses: t.Optional[bool] = None,
219 validator_map: t.Optional[dict] = None,
220 **kwargs,
221 ) -> t.Any:
222 """
223 Register een API represented by a single OpenAPI specification on this middleware.
224 Multiple APIs can be registered on a single middleware.
226 :param specification: OpenAPI specification. Can be provided either as dict, or as path
227 to file.
228 :param base_path: Base path to host the API. This overrides the basePath / servers in the
229 specification.
230 :param arguments: Arguments to substitute the specification using Jinja.
231 :param auth_all_paths: whether to authenticate not paths not defined in the specification.
232 Defaults to False.
233 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses.
234 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an
235 underscore is appended to any shadowed built-ins. Defaults to False.
236 :param resolver: Callable that maps operationId to a function or instance of
237 :class:`resolver.Resolver`.
238 :param resolver_error: Error code to return for operations for which the operationId could
239 not be resolved. If no error code is provided, the application will fail when trying to
240 start.
241 :param strict_validation: When True, extra form or query parameters not defined in the
242 specification result in a validation error. Defaults to False.
243 :param swagger_ui_options: A :class:`options.ConnexionOptions` instance with configuration
244 options for the swagger ui.
245 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`.
246 :param validate_responses: Whether to validate responses against the specification. This has
247 an impact on performance. Defaults to False.
248 :param validator_map: A dictionary of validators to use. Defaults to
249 :obj:`validators.VALIDATOR_MAP`
250 :param kwargs: Additional keyword arguments to pass to the `add_api` method of the managed
251 middlewares. This can be used to pass arguments to middlewares added beyond the default
252 ones.
254 :return: The Api registered on the wrapped application.
255 """
256 if isinstance(specification, dict):
257 specification = specification
258 else:
259 specification = t.cast(pathlib.Path, self.specification_dir / specification)
260 # Add specification as file to watch for reloading
261 if pathlib.Path.cwd() in specification.parents:
262 self.extra_files.append(
263 str(specification.relative_to(pathlib.Path.cwd()))
264 )
266 options = self.options.replace(
267 arguments=arguments,
268 auth_all_paths=auth_all_paths,
269 jsonifier=jsonifier,
270 swagger_ui_options=swagger_ui_options,
271 pythonic_params=pythonic_params,
272 resolver=resolver,
273 resolver_error=resolver_error,
274 strict_validation=strict_validation,
275 uri_parser_class=uri_parser_class,
276 validate_responses=validate_responses,
277 validator_map=validator_map,
278 )
280 for app in self.apps:
281 if isinstance(app, SpecMiddleware):
282 api = app.add_api(
283 specification,
284 base_path=base_path,
285 **options.__dict__,
286 **kwargs,
287 )
289 # Api registered on the inner application.
290 return api
292 def add_error_handler(
293 self, code_or_exception: t.Union[int, t.Type[Exception]], function: t.Callable
294 ) -> None:
295 for app in self.apps:
296 if isinstance(app, ExceptionMiddleware):
297 app.add_exception_handler(code_or_exception, function)
299 def run(self, import_string: str = None, **kwargs):
300 """Run the application using uvicorn.
302 :param import_string: application as import string (eg. "main:app"). This is needed to run
303 using reload.
304 :param kwargs: kwargs to pass to `uvicorn.run`.
305 """
306 try:
307 import uvicorn
308 except ImportError:
309 raise RuntimeError(
310 "uvicorn is not installed. Please install connexion using the uvicorn extra "
311 "(connexion[uvicorn])"
312 )
314 logger.warning(
315 f"`{self.__class__.__name__}.run` is optimized for development. "
316 "For production, run using a dedicated ASGI server."
317 )
319 app: t.Union[str, ConnexionMiddleware]
320 if import_string is not None:
321 app = import_string
322 kwargs.setdefault("reload", True)
323 kwargs["reload_includes"] = self.extra_files + kwargs.get(
324 "reload_includes", []
325 )
326 else:
327 app = self
329 uvicorn.run(app, **kwargs)
331 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
332 await self.app(scope, receive, send)