-
-
Notifications
You must be signed in to change notification settings - Fork 762
/
flask_api.py
316 lines (264 loc) · 12.7 KB
/
flask_api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
import logging
import warnings
import flask
import werkzeug.exceptions
from connexion.apis import flask_utils
from connexion.apis.abstract import AbstractAPI
from connexion.handlers import AuthErrorHandler
from connexion.jsonifier import Jsonifier
from connexion.lifecycle import ConnexionRequest, ConnexionResponse
from connexion.utils import is_json_mimetype, yamldumper
from connexion.security import FlaskSecurityHandlerFactory
from werkzeug.local import LocalProxy
logger = logging.getLogger('connexion.apis.flask_api')
class FlaskApi(AbstractAPI):
@staticmethod
def make_security_handler_factory(pass_context_arg_name):
""" Create default SecurityHandlerFactory to create all security check handlers """
return FlaskSecurityHandlerFactory(pass_context_arg_name)
def _set_base_path(self, base_path):
super(FlaskApi, self)._set_base_path(base_path)
self._set_blueprint()
def _set_blueprint(self):
logger.debug('Creating API blueprint: %s', self.base_path)
endpoint = flask_utils.flaskify_endpoint(self.base_path)
self.blueprint = flask.Blueprint(endpoint, __name__, url_prefix=self.base_path,
template_folder=str(self.options.openapi_console_ui_from_dir))
def add_openapi_json(self):
"""
Adds spec json to {base_path}/swagger.json
or {base_path}/openapi.json (for oas3)
"""
logger.debug('Adding spec json: %s/%s', self.base_path,
self.options.openapi_spec_path)
endpoint_name = "{name}_openapi_json".format(name=self.blueprint.name)
self.blueprint.add_url_rule(self.options.openapi_spec_path,
endpoint_name,
self._handlers.get_json_spec)
def add_openapi_yaml(self):
"""
Adds spec yaml to {base_path}/swagger.yaml
or {base_path}/openapi.yaml (for oas3)
"""
if not self.options.openapi_spec_path.endswith("json"):
return
openapi_spec_path_yaml = \
self.options.openapi_spec_path[:-len("json")] + "yaml"
logger.debug('Adding spec yaml: %s/%s', self.base_path,
openapi_spec_path_yaml)
endpoint_name = "{name}_openapi_yaml".format(name=self.blueprint.name)
self.blueprint.add_url_rule(
openapi_spec_path_yaml,
endpoint_name,
self._handlers.get_yaml_spec
)
def add_swagger_ui(self):
"""
Adds swagger ui to {base_path}/ui/
"""
console_ui_path = self.options.openapi_console_ui_path.strip('/')
logger.debug('Adding swagger-ui: %s/%s/',
self.base_path,
console_ui_path)
if self.options.openapi_console_ui_config is not None:
config_endpoint_name = "{name}_swagger_ui_config".format(name=self.blueprint.name)
config_file_url = '/{console_ui_path}/swagger-ui-config.json'.format(
console_ui_path=console_ui_path)
self.blueprint.add_url_rule(config_file_url,
config_endpoint_name,
lambda: flask.jsonify(self.options.openapi_console_ui_config))
static_endpoint_name = "{name}_swagger_ui_static".format(name=self.blueprint.name)
static_files_url = '/{console_ui_path}/<path:filename>'.format(
console_ui_path=console_ui_path)
self.blueprint.add_url_rule(static_files_url,
static_endpoint_name,
self._handlers.console_ui_static_files)
index_endpoint_name = "{name}_swagger_ui_index".format(name=self.blueprint.name)
console_ui_url = '/{console_ui_path}/'.format(
console_ui_path=console_ui_path)
self.blueprint.add_url_rule(console_ui_url,
index_endpoint_name,
self._handlers.console_ui_home)
def add_auth_on_not_found(self, security, security_definitions):
"""
Adds a 404 error handler to authenticate and only expose the 404 status if the security validation pass.
"""
logger.debug('Adding path not found authentication')
not_found_error = AuthErrorHandler(self, werkzeug.exceptions.NotFound(), security=security,
security_definitions=security_definitions)
endpoint_name = "{name}_not_found".format(name=self.blueprint.name)
self.blueprint.add_url_rule('/<path:invalid_path>', endpoint_name, not_found_error.function)
def _add_operation_internal(self, method, path, operation):
operation_id = operation.operation_id
logger.debug('... Adding %s -> %s', method.upper(), operation_id,
extra=vars(operation))
flask_path = flask_utils.flaskify_path(path, operation.get_path_parameter_types())
endpoint_name = flask_utils.flaskify_endpoint(operation.operation_id,
operation.randomize_endpoint)
function = operation.function
self.blueprint.add_url_rule(flask_path, endpoint_name, function, methods=[method])
@property
def _handlers(self):
# type: () -> InternalHandlers
if not hasattr(self, '_internal_handlers'):
self._internal_handlers = InternalHandlers(self.base_path, self.options, self.specification)
return self._internal_handlers
@classmethod
def get_response(cls, response, mimetype=None, request=None):
"""Gets ConnexionResponse instance for the operation handler
result. Status Code and Headers for response. If only body
data is returned by the endpoint function, then the status
code will be set to 200 and no headers will be added.
If the returned object is a flask.Response then it will just
pass the information needed to recreate it.
:type response: flask.Response | (flask.Response,) | (flask.Response, int) | (flask.Response, dict) | (flask.Response, int, dict)
:rtype: ConnexionResponse
"""
return cls._get_response(response, mimetype=mimetype, extra_context={"url": flask.request.url})
@classmethod
def _is_framework_response(cls, response):
""" Return True if provided response is a framework type """
return flask_utils.is_flask_response(response)
@classmethod
def _framework_to_connexion_response(cls, response, mimetype):
""" Cast framework response class to ConnexionResponse used for schema validation """
return ConnexionResponse(
status_code=response.status_code,
mimetype=response.mimetype,
content_type=response.content_type,
headers=response.headers,
body=response.get_data(),
)
@classmethod
def _connexion_to_framework_response(cls, response, mimetype, extra_context=None):
""" Cast ConnexionResponse to framework response class """
flask_response = cls._build_response(
mimetype=response.mimetype or mimetype,
content_type=response.content_type,
headers=response.headers,
status_code=response.status_code,
data=response.body,
extra_context=extra_context,
)
return flask_response
@classmethod
def _build_response(cls, mimetype, content_type=None, headers=None, status_code=None, data=None, extra_context=None):
if cls._is_framework_response(data):
return flask.current_app.make_response((data, status_code, headers))
data, status_code, serialized_mimetype = cls._prepare_body_and_status_code(data=data, mimetype=mimetype, status_code=status_code, extra_context=extra_context)
kwargs = {
'mimetype': mimetype or serialized_mimetype,
'content_type': content_type,
'headers': headers,
'response': data,
'status': status_code
}
kwargs = {k: v for k, v in kwargs.items() if v is not None}
return flask.current_app.response_class(**kwargs) # type: flask.Response
@classmethod
def _serialize_data(cls, data, mimetype):
# TODO: harmonize flask and aiohttp serialization when mimetype=None or mimetype is not JSON
# (cases where it might not make sense to jsonify the data)
if (isinstance(mimetype, str) and is_json_mimetype(mimetype)):
body = cls.jsonifier.dumps(data)
elif not (isinstance(data, bytes) or isinstance(data, str)):
warnings.warn(
"Implicit (flask) JSON serialization will change in the next major version. "
"This is triggered because a response body is being serialized as JSON "
"even though the mimetype is not a JSON type. "
"This will be replaced by something that is mimetype-specific and may "
"raise an error instead of silently converting everything to JSON. "
"Please make sure to specify media/mime types in your specs.",
FutureWarning # a Deprecation targeted at application users.
)
body = cls.jsonifier.dumps(data)
else:
body = data
return body, mimetype
@classmethod
def get_request(cls, *args, **params):
# type: (*Any, **Any) -> ConnexionRequest
"""Gets ConnexionRequest instance for the operation handler
result. Status Code and Headers for response. If only body
data is returned by the endpoint function, then the status
code will be set to 200 and no headers will be added.
If the returned object is a flask.Response then it will just
pass the information needed to recreate it.
:rtype: ConnexionRequest
"""
context_dict = {}
setattr(flask._request_ctx_stack.top, 'connexion_context', context_dict)
flask_request = flask.request
request = ConnexionRequest(
flask_request.url,
flask_request.method,
headers=flask_request.headers,
form=flask_request.form,
query=flask_request.args,
body=flask_request.get_data(),
json_getter=lambda: flask_request.get_json(silent=True),
files=flask_request.files,
path_params=params,
context=context_dict
)
logger.debug('Getting data and status code',
extra={
'data': request.body,
'data_type': type(request.body),
'url': request.url
})
return request
@classmethod
def _set_jsonifier(cls):
"""
Use Flask specific JSON loader
"""
cls.jsonifier = Jsonifier(flask.json, indent=2)
def _get_context():
return getattr(flask._request_ctx_stack.top, 'connexion_context')
context = LocalProxy(_get_context)
class InternalHandlers(object):
"""
Flask handlers for internally registered endpoints.
"""
def __init__(self, base_path, options, specification):
self.base_path = base_path
self.options = options
self.specification = specification
def console_ui_home(self):
"""
Home page of the OpenAPI Console UI.
:return:
"""
openapi_json_route_name = "{blueprint}.{prefix}_openapi_json"
escaped = flask_utils.flaskify_endpoint(self.base_path)
openapi_json_route_name = openapi_json_route_name.format(
blueprint=escaped,
prefix=escaped
)
template_variables = {
'openapi_spec_url': flask.url_for(openapi_json_route_name)
}
if self.options.openapi_console_ui_config is not None:
template_variables['configUrl'] = 'swagger-ui-config.json'
return flask.render_template('index.j2', **template_variables)
def console_ui_static_files(self, filename):
"""
Servers the static files for the OpenAPI Console UI.
:param filename: Requested file contents.
:return:
"""
# convert PosixPath to str
static_dir = str(self.options.openapi_console_ui_from_dir)
return flask.send_from_directory(static_dir, filename)
def get_json_spec(self):
return flask.jsonify(self._spec_for_prefix())
def get_yaml_spec(self):
return yamldumper(self._spec_for_prefix()), 200, {"Content-Type": "text/yaml"}
def _spec_for_prefix(self):
"""
Modify base_path in the spec based on incoming url
This fixes problems with reverse proxies changing the path.
"""
base_path = flask.url_for(flask.request.endpoint).rsplit("/", 1)[0]
return self.specification.with_base_path(base_path).raw