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
1"""
2This module defines an AbstractApp, which defines a standardized user interface for a Connexion
3application.
4"""
5import abc
6import pathlib
7import typing as t
9from starlette.testclient import TestClient
10from starlette.types import ASGIApp, Receive, Scope, Send
12from connexion.jsonifier import Jsonifier
13from connexion.lifecycle import ConnexionRequest, ConnexionResponse
14from connexion.middleware import ConnexionMiddleware, MiddlewarePosition, SpecMiddleware
15from connexion.middleware.lifespan import Lifespan
16from connexion.options import SwaggerUIOptions
17from connexion.resolver import Resolver
18from connexion.types import MaybeAwaitable
19from connexion.uri_parsing import AbstractURIParser
22class AbstractApp:
23 """
24 Abstract class for a Connexion Application. A Connexion Application provides an interface for a
25 framework application wrapped by Connexion Middleware. Since its main function is to provide an
26 interface, it delegates most of the work to the middleware and framework application.
27 """
29 _middleware_app: SpecMiddleware
30 """
31 The application wrapped by the ConnexionMiddleware, which in its turn wraps the framework
32 application.
33 """
35 def __init__(
36 self,
37 import_name: str,
38 *,
39 lifespan: t.Optional[Lifespan] = None,
40 middlewares: t.Optional[list] = None,
41 specification_dir: t.Union[pathlib.Path, str] = "",
42 arguments: t.Optional[dict] = None,
43 auth_all_paths: t.Optional[bool] = None,
44 jsonifier: t.Optional[Jsonifier] = None,
45 pythonic_params: t.Optional[bool] = None,
46 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None,
47 resolver_error: t.Optional[int] = None,
48 strict_validation: t.Optional[bool] = None,
49 swagger_ui_options: t.Optional[SwaggerUIOptions] = None,
50 uri_parser_class: t.Optional[AbstractURIParser] = None,
51 validate_responses: t.Optional[bool] = None,
52 validator_map: t.Optional[dict] = None,
53 security_map: t.Optional[dict] = None,
54 ) -> None:
55 """
56 :param import_name: The name of the package or module that this object belongs to. If you
57 are using a single module, __name__ is always the correct value. If you however are
58 using a package, it’s usually recommended to hardcode the name of your package there.
59 :param lifespan: A lifespan context function, which can be used to perform startup and
60 :param middlewares: The list of middlewares to wrap around the application. Defaults to
61 :obj:`middleware.main.ConnexionMiddleware.default_middlewares`
62 :param specification_dir: The directory holding the specification(s). The provided path
63 should either be absolute or relative to the root path of the application. Defaults to
64 the root path.
65 :param arguments: Arguments to substitute the specification using Jinja.
66 :param auth_all_paths: whether to authenticate not paths not defined in the specification.
67 Defaults to False.
68 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses.
69 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an
70 underscore is appended to any shadowed built-ins. Defaults to False.
71 :param resolver: Callable that maps operationId to a function or instance of
72 :class:`resolver.Resolver`.
73 :param resolver_error: Error code to return for operations for which the operationId could
74 not be resolved. If no error code is provided, the application will fail when trying to
75 start.
76 :param strict_validation: When True, extra form or query parameters not defined in the
77 specification result in a validation error. Defaults to False.
78 :param swagger_ui_options: Instance of :class:`options.ConnexionOptions` with
79 configuration options for the swagger ui.
80 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`.
81 :param validate_responses: Whether to validate responses against the specification. This has
82 an impact on performance. Defaults to False.
83 :param validator_map: A dictionary of validators to use. Defaults to
84 :obj:`validators.VALIDATOR_MAP`.
85 :param security_map: A dictionary of security handlers to use. Defaults to
86 :obj:`security.SECURITY_HANDLERS`
87 """
88 self.middleware = ConnexionMiddleware(
89 self._middleware_app,
90 import_name=import_name,
91 lifespan=lifespan,
92 middlewares=middlewares,
93 specification_dir=specification_dir,
94 arguments=arguments,
95 auth_all_paths=auth_all_paths,
96 jsonifier=jsonifier,
97 swagger_ui_options=swagger_ui_options,
98 pythonic_params=pythonic_params,
99 resolver=resolver,
100 resolver_error=resolver_error,
101 strict_validation=strict_validation,
102 uri_parser_class=uri_parser_class,
103 validate_responses=validate_responses,
104 validator_map=validator_map,
105 security_map=security_map,
106 )
108 def add_middleware(
109 self,
110 middleware_class: t.Type[ASGIApp],
111 position: MiddlewarePosition = MiddlewarePosition.BEFORE_CONTEXT,
112 **options: t.Any,
113 ) -> None:
114 """Add a middleware to the stack on the specified position.
116 :param middleware_class: Middleware class to add
117 :param position: Position to add the middleware, one of the MiddlewarePosition Enum
118 :param options: Options to pass to the middleware_class on initialization
119 """
120 self.middleware.add_middleware(middleware_class, position=position, **options)
122 def add_api(
123 self,
124 specification: t.Union[pathlib.Path, str, dict],
125 *,
126 base_path: t.Optional[str] = None,
127 name: t.Optional[str] = None,
128 arguments: t.Optional[dict] = None,
129 auth_all_paths: t.Optional[bool] = None,
130 jsonifier: t.Optional[Jsonifier] = None,
131 pythonic_params: t.Optional[bool] = None,
132 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None,
133 resolver_error: t.Optional[int] = None,
134 strict_validation: t.Optional[bool] = None,
135 swagger_ui_options: t.Optional[SwaggerUIOptions] = None,
136 uri_parser_class: t.Optional[AbstractURIParser] = None,
137 validate_responses: t.Optional[bool] = None,
138 validator_map: t.Optional[dict] = None,
139 security_map: t.Optional[dict] = None,
140 **kwargs,
141 ) -> t.Any:
142 """
143 Register an API represented by a single OpenAPI specification on this application.
144 Multiple APIs can be registered on a single application.
146 :param specification: OpenAPI specification. Can be provided either as dict, a path
147 to file, or a URL.
148 :param base_path: Base path to host the API. This overrides the basePath / servers in the
149 specification.
150 :param name: Name to register the API with. If no name is passed, the base_path is used
151 as name instead.
152 :param arguments: Arguments to substitute the specification using Jinja.
153 :param auth_all_paths: whether to authenticate not paths not defined in the specification.
154 Defaults to False.
155 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses.
156 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an
157 underscore is appended to any shadowed built-ins. Defaults to False.
158 :param resolver: Callable that maps operationId to a function or instance of
159 :class:`resolver.Resolver`.
160 :param resolver_error: Error code to return for operations for which the operationId could
161 not be resolved. If no error code is provided, the application will fail when trying to
162 start.
163 :param strict_validation: When True, extra form or query parameters not defined in the
164 specification result in a validation error. Defaults to False.
165 :param swagger_ui_options: A :class:`options.ConnexionOptions` instance with configuration
166 options for the swagger ui.
167 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`.
168 :param validate_responses: Whether to validate responses against the specification. This has
169 an impact on performance. Defaults to False.
170 :param validator_map: A dictionary of validators to use. Defaults to
171 :obj:`validators.VALIDATOR_MAP`
172 :param security_map: A dictionary of security handlers to use. Defaults to
173 :obj:`security.SECURITY_HANDLERS`
174 :param kwargs: Additional keyword arguments to pass to the `add_api` method of the managed
175 middlewares. This can be used to pass arguments to middlewares added beyond the default
176 ones.
178 :return: The Api registered on the middleware application wrapping the framework.
179 """
180 return self.middleware.add_api(
181 specification,
182 base_path=base_path,
183 name=name,
184 arguments=arguments,
185 auth_all_paths=auth_all_paths,
186 jsonifier=jsonifier,
187 pythonic_params=pythonic_params,
188 resolver=resolver,
189 resolver_error=resolver_error,
190 strict_validation=strict_validation,
191 swagger_ui_options=swagger_ui_options,
192 uri_parser_class=uri_parser_class,
193 validate_responses=validate_responses,
194 validator_map=validator_map,
195 security_map=security_map,
196 **kwargs,
197 )
199 def add_url_rule(
200 self, rule, endpoint: str = None, view_func: t.Callable = None, **options
201 ):
202 """
203 Connects a URL rule. Works exactly like the `route` decorator.
205 Basically this example::
207 @app.route('/')
208 def index():
209 pass
211 Is equivalent to the following::
213 def index():
214 pass
215 app.add_url_rule('/', 'index', index)
217 Internally`route` invokes `add_url_rule` so if you want to customize the behavior via
218 subclassing you only need to change this method.
220 :param rule: the URL rule as string.
221 :param endpoint: the name of the endpoint for the registered URL rule, which is used for
222 reverse lookup. Flask defaults to the name of the view function.
223 :param view_func: the function to call when serving a request to the provided endpoint.
224 :param options: the options to be forwarded to the underlying ``werkzeug.routing.Rule``
225 object. A change to Werkzeug is handling of method options. methods is a list of
226 methods this rule should be limited to (`GET`, `POST` etc.). By default a rule just
227 listens for `GET` (and implicitly `HEAD`).
228 """
230 def route(self, rule: str, **options):
231 """
232 A decorator that is used to register a view function for a
233 given URL rule. This does the same thing as `add_url_rule`
234 but is intended for decorator usage::
236 @app.route('/')
237 def index():
238 return 'Hello World'
240 :param rule: the URL rule as string
241 :param options: the options to be forwarded to the underlying ``werkzeug.routing.Rule``
242 object. A change to Werkzeug is handling of method options. methods is a
243 list of methods this rule should be limited to (`GET`, `POST` etc.).
244 By default a rule just listens for `GET` (and implicitly `HEAD`).
245 """
247 def decorator(func: t.Callable) -> t.Callable:
248 self.add_url_rule(rule, view_func=func, **options)
249 return func
251 return decorator
253 @abc.abstractmethod
254 def add_error_handler(
255 self,
256 code_or_exception: t.Union[int, t.Type[Exception]],
257 function: t.Callable[
258 [ConnexionRequest, Exception], MaybeAwaitable[ConnexionResponse]
259 ],
260 ) -> None:
261 """
262 Register a callable to handle application errors.
264 :param code_or_exception: An exception class or the status code of HTTP exceptions to
265 handle.
266 :param function: Callable that will handle exception, may be async.
267 """
269 def test_client(self, **kwargs):
270 """Creates a test client for this application. The keywords arguments passed in are
271 passed to the ``StarletteClient``."""
272 return TestClient(self, **kwargs)
274 def run(self, import_string: str = None, **kwargs):
275 """Run the application using uvicorn.
277 :param import_string: application as import string (eg. "main:app"). This is needed to run
278 using reload.
279 :param kwargs: kwargs to pass to `uvicorn.run`.
280 """
281 self.middleware.run(import_string, **kwargs)
283 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
284 return await self.middleware(scope, receive, send)