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