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""" 

4 

5import functools 

6import pathlib 

7import typing as t 

8 

9import flask 

10import starlette.exceptions 

11import werkzeug.exceptions 

12from a2wsgi import WSGIMiddleware 

13from flask import Response as FlaskResponse 

14from starlette.types import Receive, Scope, Send 

15 

16from connexion.apps.abstract import AbstractApp 

17from connexion.decorators import FlaskDecorator 

18from connexion.exceptions import ResolverError 

19from connexion.frameworks import flask as flask_utils 

20from connexion.jsonifier import Jsonifier 

21from connexion.lifecycle import ConnexionRequest, ConnexionResponse 

22from connexion.middleware.abstract import AbstractRoutingAPI, SpecMiddleware 

23from connexion.middleware.lifespan import Lifespan 

24from connexion.operations import AbstractOperation 

25from connexion.options import SwaggerUIOptions 

26from connexion.resolver import Resolver 

27from connexion.types import MaybeAwaitable, WSGIApp 

28from connexion.uri_parsing import AbstractURIParser 

29 

30 

31class FlaskOperation: 

32 def __init__( 

33 self, 

34 fn: t.Callable, 

35 jsonifier: Jsonifier, 

36 operation_id: str, 

37 pythonic_params: bool, 

38 ) -> None: 

39 self._fn = fn 

40 self.jsonifier = jsonifier 

41 self.operation_id = operation_id 

42 self.pythonic_params = pythonic_params 

43 functools.update_wrapper(self, fn) 

44 

45 @classmethod 

46 def from_operation( 

47 cls, 

48 operation: AbstractOperation, 

49 *, 

50 pythonic_params: bool, 

51 jsonifier: Jsonifier, 

52 ) -> "FlaskOperation": 

53 return cls( 

54 fn=operation.function, 

55 jsonifier=jsonifier, 

56 operation_id=operation.operation_id, 

57 pythonic_params=pythonic_params, 

58 ) 

59 

60 @property 

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

62 decorator = FlaskDecorator( 

63 pythonic_params=self.pythonic_params, 

64 jsonifier=self.jsonifier, 

65 ) 

66 return decorator(self._fn) 

67 

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

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

70 

71 

72class FlaskApi(AbstractRoutingAPI): 

73 def __init__( 

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

75 ) -> None: 

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

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

78 

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

80 super()._set_base_path(base_path) 

81 self._set_blueprint() 

82 

83 def _set_blueprint(self): 

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

85 self.blueprint = flask.Blueprint( 

86 endpoint, 

87 __name__, 

88 url_prefix=self.base_path, 

89 ) 

90 

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

92 pass 

93 

94 def make_operation(self, operation): 

95 return FlaskOperation.from_operation( 

96 operation, pythonic_params=self.pythonic_params, jsonifier=self.jsonifier 

97 ) 

98 

99 @staticmethod 

100 def _framework_path_and_name( 

101 operation: AbstractOperation, path: str 

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

103 flask_path = flask_utils.flaskify_path( 

104 path, operation.get_path_parameter_types() 

105 ) 

106 endpoint_name = flask_utils.flaskify_endpoint( 

107 operation.operation_id, operation.randomize_endpoint 

108 ) 

109 return flask_path, endpoint_name 

110 

111 def _add_operation_internal( 

112 self, 

113 method: str, 

114 path: str, 

115 operation: t.Callable, 

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

117 ) -> None: 

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

119 

120 def add_url_rule( 

121 self, 

122 rule, 

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

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

125 **options, 

126 ): 

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

128 

129 

130class FlaskASGIApp(SpecMiddleware): 

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

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

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

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

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

136 

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

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

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

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

141 

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

143 

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

145 api = FlaskApi(specification, **kwargs) 

146 

147 if name is not None: 

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

149 else: 

150 self.app.register_blueprint(api.blueprint) 

151 

152 return api 

153 

154 def add_url_rule( 

155 self, 

156 rule, 

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

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

159 **options, 

160 ): 

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

162 

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

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

165 

166 

167class FlaskApp(AbstractApp): 

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

169 

170 _middleware_app: FlaskASGIApp 

171 

172 def __init__( 

173 self, 

174 import_name: str, 

175 *, 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

192 ): 

193 """ 

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

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

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

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

198 shutdown tasks. 

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

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

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

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

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

204 the root path. 

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

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

207 Defaults to False. 

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

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

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

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

212 :class:`resolver.Resolver`. 

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

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

215 start. 

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

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

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

219 configuration options for the swagger ui. 

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

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

222 an impact on performance. Defaults to False. 

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

224 :obj:`validators.VALIDATOR_MAP`. 

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

226 :obj:`security.SECURITY_HANDLERS` 

227 """ 

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

229 

230 super().__init__( 

231 import_name, 

232 lifespan=lifespan, 

233 middlewares=middlewares, 

234 specification_dir=specification_dir, 

235 arguments=arguments, 

236 auth_all_paths=auth_all_paths, 

237 jsonifier=jsonifier, 

238 pythonic_params=pythonic_params, 

239 resolver=resolver, 

240 resolver_error=resolver_error, 

241 strict_validation=strict_validation, 

242 swagger_ui_options=swagger_ui_options, 

243 uri_parser_class=uri_parser_class, 

244 validate_responses=validate_responses, 

245 validator_map=validator_map, 

246 security_map=security_map, 

247 ) 

248 

249 self.app = self._middleware_app.app 

250 self.app.register_error_handler( 

251 werkzeug.exceptions.HTTPException, self._http_exception 

252 ) 

253 

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

255 """Reraise werkzeug HTTPExceptions as starlette HTTPExceptions""" 

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

257 

258 def add_url_rule( 

259 self, 

260 rule, 

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

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

263 **options, 

264 ): 

265 self._middleware_app.add_url_rule( 

266 rule, endpoint=endpoint, view_func=view_func, **options 

267 ) 

268 

269 def add_error_handler( 

270 self, 

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

272 function: t.Callable[ 

273 [ConnexionRequest, Exception], MaybeAwaitable[ConnexionResponse] 

274 ], 

275 ) -> None: 

276 self.middleware.add_error_handler(code_or_exception, function) 

277 

278 def add_wsgi_middleware( 

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

280 ) -> None: 

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

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

283 be added as ASGI middleware instead. 

284 

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

286 

287 :param middleware: Middleware class to add 

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

289 """ 

290 self._middleware_app.asgi_app.app = middleware( 

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

292 )