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

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, 

112 method: str, 

113 path: str, 

114 operation: t.Callable, 

115 name: t.Optional[str] = None, 

116 ) -> None: 

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

118 

119 def add_url_rule( 

120 self, 

121 rule, 

122 endpoint: t.Optional[str] = None, 

123 view_func: t.Optional[t.Callable] = None, 

124 **options, 

125 ): 

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

127 

128 

129class FlaskASGIApp(SpecMiddleware): 

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

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

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

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

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

135 

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

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

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

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

140 

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

142 

143 def add_api(self, specification, *, name: t.Optional[str] = None, **kwargs): 

144 api = FlaskApi(specification, **kwargs) 

145 

146 if name is not None: 

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

148 else: 

149 self.app.register_blueprint(api.blueprint) 

150 

151 return api 

152 

153 def add_url_rule( 

154 self, 

155 rule, 

156 endpoint: t.Optional[str] = None, 

157 view_func: t.Optional[t.Callable] = None, 

158 **options, 

159 ): 

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

161 

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

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

164 

165 

166class FlaskApp(AbstractApp): 

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

168 

169 _middleware_app: FlaskASGIApp 

170 

171 def __init__( 

172 self, 

173 import_name: str, 

174 *, 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

191 ): 

192 """ 

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

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

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

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

197 shutdown tasks. 

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

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

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

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

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

203 the root path. 

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

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

206 Defaults to False. 

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

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

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

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

211 :class:`resolver.Resolver`. 

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

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

214 start. 

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

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

217 :param swagger_ui_options: Instance of :class:`options.SwaggerUIOptions` with 

218 configuration options for the swagger ui. 

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

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

221 an impact on performance. Defaults to False. 

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

223 :obj:`validators.VALIDATOR_MAP`. 

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

225 :obj:`security.SECURITY_HANDLERS` 

226 """ 

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

228 

229 super().__init__( 

230 import_name, 

231 lifespan=lifespan, 

232 middlewares=middlewares, 

233 specification_dir=specification_dir, 

234 arguments=arguments, 

235 auth_all_paths=auth_all_paths, 

236 jsonifier=jsonifier, 

237 pythonic_params=pythonic_params, 

238 resolver=resolver, 

239 resolver_error=resolver_error, 

240 strict_validation=strict_validation, 

241 swagger_ui_options=swagger_ui_options, 

242 uri_parser_class=uri_parser_class, 

243 validate_responses=validate_responses, 

244 validator_map=validator_map, 

245 security_map=security_map, 

246 ) 

247 

248 self.app = self._middleware_app.app 

249 self.app.register_error_handler( 

250 werkzeug.exceptions.HTTPException, self._http_exception 

251 ) 

252 

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

254 """Reraise werkzeug HTTPExceptions as starlette HTTPExceptions""" 

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

256 

257 def add_url_rule( 

258 self, 

259 rule, 

260 endpoint: t.Optional[str] = None, 

261 view_func: t.Optional[t.Callable] = None, 

262 **options, 

263 ): 

264 self._middleware_app.add_url_rule( 

265 rule, endpoint=endpoint, view_func=view_func, **options 

266 ) 

267 

268 def add_error_handler( 

269 self, 

270 code_or_exception: t.Union[int, t.Type[Exception]], 

271 function: t.Callable[ 

272 [ConnexionRequest, Exception], MaybeAwaitable[ConnexionResponse] 

273 ], 

274 ) -> None: 

275 self.middleware.add_error_handler(code_or_exception, function) 

276 

277 def add_wsgi_middleware( 

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

279 ) -> None: 

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

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

282 be added as ASGI middleware instead. 

283 

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

285 

286 :param middleware: Middleware class to add 

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

288 """ 

289 self._middleware_app.asgi_app.app = middleware( 

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

291 )