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

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 werkzeug.exceptions 

10from a2wsgi import WSGIMiddleware 

11from flask import Response as FlaskResponse 

12from flask import signals 

13from starlette.types import Receive, Scope, Send 

14 

15from connexion.apps.abstract import AbstractApp 

16from connexion.decorators import FlaskDecorator 

17from connexion.exceptions import InternalServerError, ProblemException, ResolverError 

18from connexion.frameworks import flask as flask_utils 

19from connexion.jsonifier import Jsonifier 

20from connexion.middleware.abstract import AbstractRoutingAPI, SpecMiddleware 

21from connexion.middleware.lifespan import Lifespan 

22from connexion.operations import AbstractOperation 

23from connexion.problem import problem 

24from connexion.resolver import Resolver 

25from connexion.uri_parsing import AbstractURIParser 

26 

27 

28class FlaskOperation: 

29 def __init__( 

30 self, 

31 fn: t.Callable, 

32 jsonifier: Jsonifier, 

33 operation_id: str, 

34 pythonic_params: bool, 

35 ) -> None: 

36 self._fn = fn 

37 self.jsonifier = jsonifier 

38 self.operation_id = operation_id 

39 self.pythonic_params = pythonic_params 

40 functools.update_wrapper(self, fn) 

41 

42 @classmethod 

43 def from_operation( 

44 cls, 

45 operation: AbstractOperation, 

46 *, 

47 pythonic_params: bool, 

48 jsonifier: Jsonifier, 

49 ) -> "FlaskOperation": 

50 return cls( 

51 fn=operation.function, 

52 jsonifier=jsonifier, 

53 operation_id=operation.operation_id, 

54 pythonic_params=pythonic_params, 

55 ) 

56 

57 @property 

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

59 decorator = FlaskDecorator( 

60 pythonic_params=self.pythonic_params, 

61 jsonifier=self.jsonifier, 

62 ) 

63 return decorator(self._fn) 

64 

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

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

67 

68 

69class FlaskApi(AbstractRoutingAPI): 

70 def __init__( 

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

72 ) -> None: 

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

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

75 

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

77 super()._set_base_path(base_path) 

78 self._set_blueprint() 

79 

80 def _set_blueprint(self): 

81 endpoint = flask_utils.flaskify_endpoint(self.base_path) 

82 self.blueprint = flask.Blueprint( 

83 endpoint, 

84 __name__, 

85 url_prefix=self.base_path, 

86 ) 

87 

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

89 pass 

90 

91 def make_operation(self, operation): 

92 return FlaskOperation.from_operation( 

93 operation, pythonic_params=self.pythonic_params, jsonifier=self.jsonifier 

94 ) 

95 

96 @staticmethod 

97 def _framework_path_and_name( 

98 operation: AbstractOperation, path: str 

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

100 flask_path = flask_utils.flaskify_path( 

101 path, operation.get_path_parameter_types() 

102 ) 

103 endpoint_name = flask_utils.flaskify_endpoint( 

104 operation.operation_id, operation.randomize_endpoint 

105 ) 

106 return flask_path, endpoint_name 

107 

108 def _add_operation_internal( 

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

110 ) -> None: 

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

112 

113 def add_url_rule( 

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

115 ): 

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

117 

118 

119class FlaskMiddlewareApp(SpecMiddleware): 

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

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

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

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

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

125 

126 self.set_errors_handlers() 

127 

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

129 

130 def set_errors_handlers(self): 

131 for error_code in werkzeug.exceptions.default_exceptions: 

132 self.app.register_error_handler(error_code, self.common_error_handler) 

133 

134 self.app.register_error_handler(ProblemException, self.common_error_handler) 

135 

136 def common_error_handler(self, exception: Exception) -> FlaskResponse: 

137 """Default error handler.""" 

138 if isinstance(exception, ProblemException): 

139 response = exception.to_problem() 

140 else: 

141 if not isinstance(exception, werkzeug.exceptions.HTTPException): 

142 exception = InternalServerError() 

143 

144 response = problem( 

145 title=exception.name, 

146 detail=exception.description, 

147 status=exception.code, 

148 ) 

149 

150 if response.status_code >= 500: 

151 signals.got_request_exception.send(self.app, exception=exception) 

152 

153 return flask.make_response( 

154 (response.body, response.status_code, response.headers) 

155 ) 

156 

157 def add_api(self, specification, **kwargs): 

158 api = FlaskApi(specification, **kwargs) 

159 self.app.register_blueprint(api.blueprint) 

160 return api 

161 

162 def add_url_rule( 

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

164 ): 

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

166 

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

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

169 

170 

171class FlaskApp(AbstractApp): 

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

173 

174 middleware_app: FlaskMiddlewareApp 

175 

176 def __init__( 

177 self, 

178 import_name: str, 

179 *, 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

195 ): 

196 """ 

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

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

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

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

201 shutdown tasks. 

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

203 :obj:`middleware.main.ConnexionmMiddleware.default_middlewares` 

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

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

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

207 the root path. 

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

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

210 Defaults to False. 

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

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

213 options for the swagger ui. 

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

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

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

217 :class:`resolver.Resolver`. 

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

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

220 start. 

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

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

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

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

225 an impact on performance. Defaults to False. 

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

227 :obj:`validators.VALIDATOR_MAP`. 

228 """ 

229 self.middleware_app = FlaskMiddlewareApp(import_name, server_args or {}) 

230 self.app = self.middleware_app.app 

231 super().__init__( 

232 import_name, 

233 lifespan=lifespan, 

234 middlewares=middlewares, 

235 specification_dir=specification_dir, 

236 arguments=arguments, 

237 auth_all_paths=auth_all_paths, 

238 jsonifier=jsonifier, 

239 pythonic_params=pythonic_params, 

240 resolver=resolver, 

241 resolver_error=resolver_error, 

242 strict_validation=strict_validation, 

243 swagger_ui_options=swagger_ui_options, 

244 uri_parser_class=uri_parser_class, 

245 validate_responses=validate_responses, 

246 validator_map=validator_map, 

247 ) 

248 

249 def add_url_rule( 

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

251 ): 

252 self.middleware_app.add_url_rule( 

253 rule, endpoint=endpoint, view_func=view_func, **options 

254 ) 

255 

256 def add_error_handler( 

257 self, code_or_exception: t.Union[int, t.Type[Exception]], function: t.Callable 

258 ) -> None: 

259 self.app.register_error_handler(code_or_exception, function)