Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/apps/flask.py: 7%

1""" 

2This module defines a FlaskApp, a Connexion application to wrap a Flask application. 

3""" 

4import functools 

5import pathlib 

6import typing as t 

7 

8import flask 

9import starlette.exceptions 

10import werkzeug.exceptions 

11from a2wsgi import WSGIMiddleware 

12from flask import Response as FlaskResponse 

13from starlette.types import Receive, Scope, Send 

14 

15from connexion.apps.abstract import AbstractApp 

16from connexion.decorators import FlaskDecorator 

17from connexion.exceptions import ResolverError 

18from connexion.frameworks import flask as flask_utils 

19from connexion.jsonifier import Jsonifier 

20from connexion.lifecycle import ConnexionRequest, ConnexionResponse 

21from connexion.middleware.abstract import AbstractRoutingAPI, SpecMiddleware 

22from connexion.middleware.lifespan import Lifespan 

23from connexion.operations import AbstractOperation 

24from connexion.options import SwaggerUIOptions 

25from connexion.resolver import Resolver 

26from connexion.types import MaybeAwaitable, WSGIApp 

27from connexion.uri_parsing import AbstractURIParser 

28 

29 

30class FlaskOperation: 

31 def __init__( 

32 self, 

33 fn: t.Callable, 

34 jsonifier: Jsonifier, 

35 operation_id: str, 

36 pythonic_params: bool, 

37 ) -> None: 

38 self._fn = fn 

39 self.jsonifier = jsonifier 

40 self.operation_id = operation_id 

41 self.pythonic_params = pythonic_params 

42 functools.update_wrapper(self, fn) 

43 

44 @classmethod 

45 def from_operation( 

46 cls, 

47 operation: AbstractOperation, 

48 *, 

49 pythonic_params: bool, 

50 jsonifier: Jsonifier, 

51 ) -> "FlaskOperation": 

52 return cls( 

53 fn=operation.function, 

54 jsonifier=jsonifier, 

55 operation_id=operation.operation_id, 

56 pythonic_params=pythonic_params, 

57 ) 

58 

59 @property 

60 def fn(self) -> t.Callable: 

61 decorator = FlaskDecorator( 

62 pythonic_params=self.pythonic_params, 

63 jsonifier=self.jsonifier, 

64 ) 

65 return decorator(self._fn) 

66 

67 def __call__(self, *args, **kwargs) -> FlaskResponse: 

68 return self.fn(*args, **kwargs) 

69 

70 

71class FlaskApi(AbstractRoutingAPI): 

72 def __init__( 

73 self, *args, jsonifier: t.Optional[Jsonifier] = None, **kwargs 

74 ) -> None: 

75 self.jsonifier = jsonifier or Jsonifier(flask.json, indent=2) 

76 super().__init__(*args, **kwargs) 

77 

78 def _set_base_path(self, base_path: t.Optional[str] = None) -> None: 

79 super()._set_base_path(base_path) 

80 self._set_blueprint() 

81 

82 def _set_blueprint(self): 

83 endpoint = flask_utils.flaskify_endpoint(self.base_path) or "/" 

84 self.blueprint = flask.Blueprint( 

85 endpoint, 

86 __name__, 

87 url_prefix=self.base_path, 

88 ) 

89 

90 def _add_resolver_error_handler(self, method: str, path: str, err: ResolverError): 

91 pass 

92 

93 def make_operation(self, operation): 

94 return FlaskOperation.from_operation( 

95 operation, pythonic_params=self.pythonic_params, jsonifier=self.jsonifier 

96 ) 

97 

98 @staticmethod 

99 def _framework_path_and_name( 

100 operation: AbstractOperation, path: str 

101 ) -> t.Tuple[str, str]: 

102 flask_path = flask_utils.flaskify_path( 

103 path, operation.get_path_parameter_types() 

104 ) 

105 endpoint_name = flask_utils.flaskify_endpoint( 

106 operation.operation_id, operation.randomize_endpoint 

107 ) 

108 return flask_path, endpoint_name 

109 

110 def _add_operation_internal( 

111 self, method: str, path: str, operation: t.Callable, name: str = None 

112 ) -> None: 

113 self.blueprint.add_url_rule(path, name, operation, methods=[method]) 

114 

115 def add_url_rule( 

116 self, rule, endpoint: str = None, view_func: t.Callable = None, **options 

117 ): 

118 return self.blueprint.add_url_rule(rule, endpoint, view_func, **options) 

119 

120 

121class FlaskASGIApp(SpecMiddleware): 

122 def __init__(self, import_name, server_args: dict, **kwargs): 

123 self.app = flask.Flask(import_name, **server_args) 

124 self.app.json = flask_utils.FlaskJSONProvider(self.app) 

125 self.app.url_map.converters["float"] = flask_utils.NumberConverter 

126 self.app.url_map.converters["int"] = flask_utils.IntegerConverter 

127 

128 # Propagate Errors so we can handle them in the middleware 

129 self.app.config["PROPAGATE_EXCEPTIONS"] = True 

130 self.app.config["TRAP_BAD_REQUEST_ERRORS"] = True 

131 self.app.config["TRAP_HTTP_EXCEPTIONS"] = True 

132 

133 self.asgi_app = WSGIMiddleware(self.app.wsgi_app) 

134 

135 def add_api(self, specification, *, name: str = None, **kwargs): 

136 api = FlaskApi(specification, **kwargs) 

137 

138 if name is not None: 

139 self.app.register_blueprint(api.blueprint, name=name) 

140 else: 

141 self.app.register_blueprint(api.blueprint) 

142 

143 return api 

144 

145 def add_url_rule( 

146 self, rule, endpoint: str = None, view_func: t.Callable = None, **options 

147 ): 

148 return self.app.add_url_rule(rule, endpoint, view_func, **options) 

149 

150 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

151 return await self.asgi_app(scope, receive, send) 

152 

153 

154class FlaskApp(AbstractApp): 

155 """Connexion Application based on ConnexionMiddleware wrapping a Flask application.""" 

156 

157 _middleware_app: FlaskASGIApp 

158 

159 def __init__( 

160 self, 

161 import_name: str, 

162 *, 

163 lifespan: t.Optional[Lifespan] = None, 

164 middlewares: t.Optional[list] = None, 

165 server_args: t.Optional[dict] = None, 

166 specification_dir: t.Union[pathlib.Path, str] = "", 

167 arguments: t.Optional[dict] = None, 

168 auth_all_paths: t.Optional[bool] = None, 

169 jsonifier: t.Optional[Jsonifier] = None, 

170 pythonic_params: t.Optional[bool] = None, 

171 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None, 

172 resolver_error: t.Optional[int] = None, 

173 strict_validation: t.Optional[bool] = None, 

174 swagger_ui_options: t.Optional[SwaggerUIOptions] = None, 

175 uri_parser_class: t.Optional[AbstractURIParser] = None, 

176 validate_responses: t.Optional[bool] = None, 

177 validator_map: t.Optional[dict] = None, 

178 security_map: t.Optional[dict] = None, 

179 ): 

180 """ 

181 :param import_name: The name of the package or module that this object belongs to. If you 

182 are using a single module, __name__ is always the correct value. If you however are 

183 using a package, it’s usually recommended to hardcode the name of your package there. 

184 :param lifespan: A lifespan context function, which can be used to perform startup and 

185 shutdown tasks. 

186 :param middlewares: The list of middlewares to wrap around the application. Defaults to 

187 :obj:`middleware.main.ConnexionMiddleware.default_middlewares` 

188 :param server_args: Arguments to pass to the Flask application. 

189 :param specification_dir: The directory holding the specification(s). The provided path 

190 should either be absolute or relative to the root path of the application. Defaults to 

191 the root path. 

192 :param arguments: Arguments to substitute the specification using Jinja. 

193 :param auth_all_paths: whether to authenticate not paths not defined in the specification. 

194 Defaults to False. 

195 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses. 

196 :param swagger_ui_options: A :class:`options.ConnexionOptions` instance with configuration 

197 options for the swagger ui. 

198 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an 

199 underscore is appended to any shadowed built-ins. Defaults to False. 

200 :param resolver: Callable that maps operationId to a function or instance of 

201 :class:`resolver.Resolver`. 

202 :param resolver_error: Error code to return for operations for which the operationId could 

203 not be resolved. If no error code is provided, the application will fail when trying to 

204 start. 

205 :param strict_validation: When True, extra form or query parameters not defined in the 

206 specification result in a validation error. Defaults to False. 

207 :param swagger_ui_options: Instance of :class:`options.ConnexionOptions` with 

208 configuration options for the swagger ui. 

209 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`. 

210 :param validate_responses: Whether to validate responses against the specification. This has 

211 an impact on performance. Defaults to False. 

212 :param validator_map: A dictionary of validators to use. Defaults to 

213 :obj:`validators.VALIDATOR_MAP`. 

214 :param security_map: A dictionary of security handlers to use. Defaults to 

215 :obj:`security.SECURITY_HANDLERS` 

216 """ 

217 self._middleware_app = FlaskASGIApp(import_name, server_args or {}) 

218 

219 super().__init__( 

220 import_name, 

221 lifespan=lifespan, 

222 middlewares=middlewares, 

223 specification_dir=specification_dir, 

224 arguments=arguments, 

225 auth_all_paths=auth_all_paths, 

226 jsonifier=jsonifier, 

227 pythonic_params=pythonic_params, 

228 resolver=resolver, 

229 resolver_error=resolver_error, 

230 strict_validation=strict_validation, 

231 swagger_ui_options=swagger_ui_options, 

232 uri_parser_class=uri_parser_class, 

233 validate_responses=validate_responses, 

234 validator_map=validator_map, 

235 security_map=security_map, 

236 ) 

237 

238 self.app = self._middleware_app.app 

239 self.app.register_error_handler( 

240 werkzeug.exceptions.HTTPException, self._http_exception 

241 ) 

242 

243 def _http_exception(self, exc: werkzeug.exceptions.HTTPException): 

244 """Reraise werkzeug HTTPExceptions as starlette HTTPExceptions""" 

245 raise starlette.exceptions.HTTPException(exc.code, detail=exc.description) 

246 

247 def add_url_rule( 

248 self, rule, endpoint: str = None, view_func: t.Callable = None, **options 

249 ): 

250 self._middleware_app.add_url_rule( 

251 rule, endpoint=endpoint, view_func=view_func, **options 

252 ) 

253 

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 self.middleware.add_error_handler(code_or_exception, function) 

262 

263 def add_wsgi_middleware( 

264 self, middleware: t.Type[WSGIApp], **options: t.Any 

265 ) -> None: 

266 """Wrap the underlying Flask application with a WSGI middleware. Note that it will only be 

267 called at the end of the middleware stack. Middleware that needs to act sooner, needs to 

268 be added as ASGI middleware instead. 

269 

270 Adding multiple middleware using this method wraps each middleware around the previous one. 

271 

272 :param middleware: Middleware class to add 

273 :param options: Options to pass to the middleware_class on initialization 

274 """ 

275 self._middleware_app.asgi_app.app = middleware( 

276 self._middleware_app.asgi_app.app, **options # type: ignore 

277 )