]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/dashboard/controllers/docs.py
84de247010c1bf1398073ef05f1031dde750b1d5
1 # -*- coding: utf-8 -*-
2 from __future__
import absolute_import
3 from typing
import Any
, Dict
, Union
8 from . import Controller
, BaseController
, Endpoint
, ENDPOINT_MAP
11 from ..tools
import str_to_bool
14 logger
= logging
.getLogger('controllers.docs')
17 @Controller('/docs', secure
=False)
18 class Docs(BaseController
):
21 def _gen_tags(cls
, all_endpoints
):
22 """ Generates a list of all tags and corresponding descriptions. """
23 # Scenarios to consider:
24 # * Intentionally make up a new tag name at controller => New tag name displayed.
25 # * Misspell or make up a new tag name at endpoint => Neither tag or endpoint displayed.
26 # * Misspell tag name at controller (when referring to another controller) =>
27 # Tag displayed but no endpoints assigned
28 # * Description for a tag added at multiple locations => Only one description displayed.
30 for endpoints
in ENDPOINT_MAP
.values():
31 for endpoint
in endpoints
:
32 if endpoint
.is_api
or all_endpoints
:
33 list_of_ctrl
.add(endpoint
.ctrl
)
35 tag_map
: Dict
[str, str] = {}
36 for ctrl
in list_of_ctrl
:
37 tag_name
= ctrl
.__name
__
39 if hasattr(ctrl
, 'doc_info'):
40 if ctrl
.doc_info
['tag']:
41 tag_name
= ctrl
.doc_info
['tag']
42 tag_descr
= ctrl
.doc_info
['tag_descr']
43 if tag_name
not in tag_map
or not tag_map
[tag_name
]:
44 tag_map
[tag_name
] = tag_descr
46 tags
= [{'name': k
, 'description': v
if v
else "*No description available*"}
47 for k
, v
in tag_map
.items()]
48 tags
.sort(key
=lambda e
: e
['name'])
52 def _get_tag(cls
, endpoint
):
53 """ Returns the name of a tag to assign to a path. """
57 if hasattr(func
, 'doc_info') and func
.doc_info
['tag']:
58 tag
= func
.doc_info
['tag']
59 elif hasattr(ctrl
, 'doc_info') and ctrl
.doc_info
['tag']:
60 tag
= ctrl
.doc_info
['tag']
64 def _gen_type(cls
, param
):
65 # pylint: disable=too-many-return-statements
67 Generates the type of parameter based on its name and default value,
68 using very simple heuristics.
69 Used if type is not explicitly defined.
71 param_name
= param
['name']
72 def_value
= param
['default'] if 'default' in param
else None
73 if param_name
.startswith("is_"):
75 if "size" in param_name
:
77 if "count" in param_name
:
79 if "num" in param_name
:
81 if isinstance(def_value
, bool):
83 if isinstance(def_value
, int):
88 # isinstance doesn't work: input is always <type 'type'>.
89 def _type_to_str(cls
, type_as_type
):
90 """ Used if type is explicitly defined. """
91 if type_as_type
is str:
92 type_as_str
= 'string'
93 elif type_as_type
is int:
94 type_as_str
= 'integer'
95 elif type_as_type
is bool:
96 type_as_str
= 'boolean'
97 elif type_as_type
is list or type_as_type
is tuple:
99 elif type_as_type
is float:
100 type_as_str
= 'number'
102 type_as_str
= 'object'
106 def _add_param_info(cls
, parameters
, p_info
):
108 # * Parameter name (if not nested) misspelt in decorator => parameter not displayed
109 # * Sometimes a parameter is used for several endpoints (e.g. fs_id in CephFS).
110 # Currently, there is no possibility of reuse. Should there be?
111 # But what if there are two parameters with same name but different functionality?
113 Adds explicitly described information for parameters of an endpoint.
116 * Either the parameter in p_info corresponds to an endpoint parameter. Implicit information
117 has higher priority, so only information that doesn't already exist is added.
118 * Or the parameter in p_info describes a nested parameter inside an endpoint parameter.
119 In that case there is no implicit information at all so all explicitly described info needs
124 for parameter
in parameters
:
125 if p
['name'] == parameter
['name']:
126 parameter
['type'] = p
['type']
127 parameter
['description'] = p
['description']
128 if 'nested_params' in p
:
129 parameter
['nested_params'] = cls
._add
_param
_info
([], p
['nested_params'])
134 'description': p
['description'],
135 'required': p
['required'],
138 nested_p
['default'] = p
['default']
139 if 'nested_params' in p
:
140 nested_p
['nested_params'] = cls
._add
_param
_info
([], p
['nested_params'])
141 parameters
.append(nested_p
)
146 def _gen_schema_for_content(cls
, params
):
148 Generates information to the content-object in OpenAPI Spec.
149 Used to for request body and responses.
155 if param
['required']:
156 required_params
.append(param
['name'])
160 props
['type'] = cls
._type
_to
_str
(param
['type'])
161 if 'nested_params' in param
:
162 if props
['type'] == 'array': # dict in array
163 props
['items'] = cls
._gen
_schema
_for
_content
(param
['nested_params'])
165 props
= cls
._gen
_schema
_for
_content
(param
['nested_params'])
166 elif props
['type'] == 'object': # e.g. [int]
167 props
['type'] = 'array'
168 props
['items'] = {'type': cls
._type
_to
_str
(param
['type'][0])}
170 props
['type'] = cls
._gen
_type
(param
)
171 if 'description' in param
:
172 props
['description'] = param
['description']
173 if 'default' in param
:
174 props
['default'] = param
['default']
175 properties
[param
['name']] = props
179 'properties': properties
,
182 schema
['required'] = required_params
186 def _gen_responses(cls
, method
, resp_object
=None):
187 resp
: Dict
[str, Dict
[str, Union
[str, Any
]]] = {
189 "description": "Operation exception. Please check the "
190 "response body for details."
193 "description": "Unauthenticated access. Please login first."
196 "description": "Unauthorized access. Please check your "
200 "description": "Unexpected error. Please check the "
201 "response body for the stack trace."
204 if method
.lower() == 'get':
205 resp
['200'] = {'description': "OK"}
206 if method
.lower() == 'post':
207 resp
['201'] = {'description': "Resource created."}
208 if method
.lower() == 'put':
209 resp
['200'] = {'description': "Resource updated."}
210 if method
.lower() == 'delete':
211 resp
['204'] = {'description': "Resource deleted."}
212 if method
.lower() in ['post', 'put', 'delete']:
213 resp
['202'] = {'description': "Operation is still executing."
214 " Please check the task queue."}
217 for status_code
, response_body
in resp_object
.items():
218 resp
[status_code
].update({
220 'application/json': {
221 'schema': cls
._gen
_schema
_for
_content
(response_body
)}}})
226 def _gen_params(cls
, params
, location
):
230 _type
= cls
._type
_to
_str
(param
['type'])
232 _type
= cls
._gen
_type
(param
)
233 if 'description' in param
:
234 descr
= param
['description']
236 descr
= "*No description available*"
238 'name': param
['name'],
245 if param
['required']:
246 res
['required'] = True
247 elif param
['default'] is None:
248 res
['allowEmptyValue'] = True
250 res
['default'] = param
['default']
251 parameters
.append(res
)
256 def _gen_paths(cls
, all_endpoints
):
257 method_order
= ['get', 'post', 'put', 'delete']
259 for path
, endpoints
in sorted(list(ENDPOINT_MAP
.items()),
264 endpoint_list
= sorted(endpoints
, key
=lambda e
:
265 method_order
.index(e
.method
.lower()))
266 for endpoint
in endpoint_list
:
267 if not endpoint
.is_api
and not all_endpoints
:
271 method
= endpoint
.method
274 summary
= "No description available"
277 if hasattr(func
, 'doc_info'):
278 if func
.doc_info
['summary']:
279 summary
= func
.doc_info
['summary']
280 resp
= func
.doc_info
['response']
281 p_info
= func
.doc_info
['parameters']
283 if endpoint
.path_params
:
286 cls
._add
_param
_info
(endpoint
.path_params
, p_info
), 'path'))
287 if endpoint
.query_params
:
290 cls
._add
_param
_info
(endpoint
.query_params
, p_info
), 'query'))
292 methods
[method
.lower()] = {
293 'tags': [cls
._get
_tag
(endpoint
)],
295 'description': func
.__doc
__,
296 'parameters': params
,
297 'responses': cls
._gen
_responses
(method
, resp
)
300 if method
.lower() in ['post', 'put']:
301 if endpoint
.body_params
:
302 body_params
= cls
._add
_param
_info
(endpoint
.body_params
, p_info
)
303 methods
[method
.lower()]['requestBody'] = {
305 'application/json': {
306 'schema': cls
._gen
_schema
_for
_content
(body_params
)}}}
308 if endpoint
.is_secure
:
309 methods
[method
.lower()]['security'] = [{'jwt': []}]
312 paths
[path
] = methods
316 def _gen_spec(self
, all_endpoints
=False, base_url
=""):
320 host
= cherrypy
.request
.base
321 host
= host
[host
.index(':')+3:]
322 logger
.debug("Host: %s", host
)
324 paths
= self
._gen
_paths
(all_endpoints
)
330 ssl
= str_to_bool(mgr
.get_localized_module_option('ssl', True))
337 'description': "Please note that this API is not an official "
338 "Ceph REST API to be used by third-party "
339 "applications. It's primary purpose is to serve"
340 " the requirements of the Ceph Dashboard and is"
341 " subject to change at any time. Use at your "
344 'title': "Ceph-Dashboard REST API"
347 'basePath': base_url
,
348 'servers': [{'url': "{}{}".format(cherrypy
.request
.base
, base_url
)}],
349 'tags': self
._gen
_tags
(all_endpoints
),
357 'bearerFormat': 'JWT'
365 @Endpoint(path
="api.json")
367 return self
._gen
_spec
(False, "/")
369 @Endpoint(path
="api-all.json")
370 def api_all_json(self
):
371 return self
._gen
_spec
(True, "/")
373 def _swagger_ui_page(self
, all_endpoints
=False, token
=None):
374 base
= cherrypy
.request
.base
376 spec_url
= "{}/docs/api-all.json".format(base
)
378 spec_url
= "{}/docs/api.json".format(base
)
380 auth_header
= cherrypy
.request
.headers
.get('authorization')
382 if auth_header
is not None:
383 scheme
, params
= auth_header
.split(' ', 1)
384 if scheme
.lower() == 'bearer':
387 if token
is not None:
390 api_key_callback
= """, onComplete: () => {{
391 ui.preauthorizeApiKey('jwt', '{}');
393 """.format(jwt_token
)
399 <meta charset="UTF-8">
400 <meta name="referrer" content="no-referrer" />
401 <link rel="stylesheet" type="text/css"
402 href="/swagger-ui.css" >
406 box-sizing: border-box;
407 overflow: -moz-scrollbars-vertical;
423 <div id="swagger-ui"></div>
424 <script src="/swagger-ui-bundle.js">
427 window.onload = function() {{
428 const ui = SwaggerUIBundle({{
430 dom_id: '#swagger-ui',
432 SwaggerUIBundle.presets.apis
442 """.format(spec_url
, api_key_callback
)
446 @Endpoint(json_response
=False)
447 def __call__(self
, all_endpoints
=False):
448 return self
._swagger
_ui
_page
(all_endpoints
)
450 @Endpoint('POST', path
="/", json_response
=False,
451 query_params
="{all_endpoints}")
452 def _with_token(self
, token
, all_endpoints
=False):
453 return self
._swagger
_ui
_page
(all_endpoints
, token
)