// ...
}
+
+REST API documentation
+~~~~~~~~~~~~~~~~~~~~~~
+There is an automatically generated Swagger UI page for documentation of the REST
+API endpoints.However, by default it is not very detailed. There are two
+decorators that can be used to add more information:
+
+* ``@EndpointDoc()`` for documentation of endpoints. It has four optional arguments
+ (explained below): ``description``, ``group``, ``parameters`` and``responses``.
+* ``@ControllerDoc()`` for documentation of controller or group associated with
+ the endpoints. It only takes the two first arguments: ``description`` and``group``.
+
+
+``description``: A a string with a short (1-2 sentences) description of the object.
+
+
+``group``: By default, an endpoint is grouped together with other endpoints
+within the same controller class. ``group`` is a string that can be used to
+assign an endpoint or all endpoints in a class to another controller or a
+conceived group name.
+
+
+``parameters``: A dict used to describe path, query or request body parameters.
+By default, all parameters for an endpoint are listed on the Swagger UI page,
+including information of whether the parameter is optional/required and default
+values. However, there will be no description of the parameter and the parameter
+type will only be displayed in some cases.
+When adding information, each parameters should be described as in the example
+below. Note that the parameter type should be expressed as a built-in python
+type and not as a string. Allowed values are ``str``, ``int``, ``bool``, ``float``.
+
+.. code-block:: python
+
+ @EndpointDoc(parameters={'my_string': (str, 'Description of my_string')})
+
+For body parameters, more complex cases are possible. If the parameter is a
+dictionary, the type should be replaced with a ``dict`` containing its nested
+parameters. When describing nested parameters, the same format as other
+parameters is used. However, all nested parameters are set as required by default.
+If the nested parameter is optional this must be specified as for ``item2`` in
+the example below. If a nested parameters is set to optional, it is also
+possible to specify the default value (this will not be provided automatically
+for nested parameters).
+
+.. code-block:: python
+
+ @EndpointDoc(parameters={
+ 'my_dictionary': ({
+ 'item1': (str, 'Description of item1'),
+ 'item2': (str, 'Description of item2', True), # item2 is optional
+ 'item3': (str, 'Description of item3', True, 'foo'), # item3 is optional with 'foo' as default value
+ }, 'Description of my_dictionary')})
+
+If the parameter is a ``list`` of primitive types, the type should be
+surrounded with square brackets.
+
+.. code-block:: python
+
+ @EndpointDoc(parameters={'my_list': ([int], 'Description of my_list')})
+
+If the parameter is a ``list`` with nested parameters, the nested parameters
+should be placed in a dictionary and surrounded with square brackets.
+
+.. code-block:: python
+
+ @EndpointDoc(parameters={
+ 'my_list': ([{
+ 'list_item': (str, 'Description of list_item'),
+ 'list_item2': (str, 'Description of list_item2')
+ }], 'Description of my_list')})
+
+
+``responses``: A dict used for describing responses. Rules for describing
+responses are the same as for request body parameters, with one difference:
+responses also needs to be assigned to the related response code as in the
+example below:
+
+.. code-block:: python
+
+ @EndpointDoc(responses={
+ '400':{'my_response': (str, 'Description of my_response')}
+
+
Error Handling in Python
~~~~~~~~~~~~~~~~~~~~~~~~
from ..plugins import PLUGIN_MANAGER
+def EndpointDoc(description="", group="", parameters=None, responses=None):
+ if not isinstance(description, str):
+ raise Exception("%s has been called with a description that is not a string: %s"
+ % (EndpointDoc.__name__, description))
+ elif not isinstance(group, str):
+ raise Exception("%s has been called with a groupname that is not a string: %s"
+ % (EndpointDoc.__name__, group))
+ elif parameters and not isinstance(parameters, dict):
+ raise Exception("%s has been called with parameters that is not a dict: %s"
+ % (EndpointDoc.__name__, parameters))
+ elif responses and not isinstance(responses, dict):
+ raise Exception("%s has been called with responses that is not a dict: %s"
+ % (EndpointDoc.__name__, responses))
+
+ if not parameters:
+ parameters = {}
+
+ def _split_param(name, p_type, description, optional=False, default_value=None, nested=False):
+ param = {
+ 'name': name,
+ 'description': description,
+ 'required': not optional,
+ 'nested': nested,
+ }
+ if default_value:
+ param['default'] = default_value
+ if isinstance(p_type, type):
+ param['type'] = p_type
+ else:
+ nested_params = _split_parameters(p_type, nested=True)
+ if nested_params:
+ param['type'] = type(p_type)
+ param['nested_params'] = nested_params
+ else:
+ param['type'] = p_type
+ return param
+
+ # Optional must be set to True in order to set default value and parameters format must be:
+ # 'name: (type or nested parameters, description, [optional], [default value])'
+ def _split_dict(data, nested):
+ splitted = []
+ for name, props in data.items():
+ if isinstance(name, str) and isinstance(props, tuple):
+ if len(props) == 2:
+ param = _split_param(name, props[0], props[1], nested=nested)
+ elif len(props) == 3:
+ param = _split_param(name, props[0], props[1], optional=props[2], nested=nested)
+ if len(props) == 4:
+ param = _split_param(name, props[0], props[1], props[2], props[3], nested)
+ splitted.append(param)
+ else:
+ raise Exception(
+ """Parameter %s in %s has not correct format. Valid formats are:
+ <name>: (<type>, <description>, [optional], [default value])
+ <name>: (<[type]>, <description>, [optional], [default value])
+ <name>: (<[nested parameters]>, <description>, [optional], [default value])
+ <name>: (<{nested parameters}>, <description>, [optional], [default value])"""
+ % (name, EndpointDoc.__name__))
+ return splitted
+
+ def _split_list(data, nested):
+ splitted = []
+ for item in data:
+ splitted.extend(_split_parameters(item, nested))
+ return splitted
+
+ # nested = True means parameters are inside a dict or array
+ def _split_parameters(data, nested=False):
+ param_list = []
+ if isinstance(data, dict):
+ param_list.extend(_split_dict(data, nested))
+ elif isinstance(data, (list, tuple)):
+ param_list.extend(_split_list(data, True))
+ return param_list
+
+ resp = {}
+ if responses:
+ for status_code, response_body in responses.items():
+ resp[str(status_code)] = _split_parameters(response_body)
+
+ def _wrapper(func):
+ func.doc_info = {
+ 'summary': description,
+ 'tag': group,
+ 'parameters': _split_parameters(parameters),
+ 'response': resp
+ }
+ return func
+
+ return _wrapper
+
+
+class ControllerDoc(object):
+ def __init__(self, description="", group=""):
+ self.tag = group
+ self.tag_descr = description
+
+ def __call__(self, cls):
+ cls.doc_info = {
+ 'tag': self.tag,
+ 'tag_descr': self.tag_descr
+ }
+ return cls
+
+
class Controller(object):
def __init__(self, path, base_url=None, security_scope=None, secure=True):
if security_scope and not Scope.valid_scope(security_scope):
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from distutils.util import strtobool
-
import cherrypy
from . import Controller, BaseController, Endpoint, ENDPOINT_MAP
@classmethod
def _gen_tags(cls, all_endpoints):
- ctrl_names = set()
+ """ Generates a list of all tags and corresponding descriptions. """
+ # Scenarios to consider:
+ # * Intentionally make up a new tag name at controller => New tag name displayed.
+ # * Misspell or make up a new tag name at endpoint => Neither tag or endpoint displayed.
+ # * Misspell tag name at controller (when refering to another controller) =>
+ # Tag displayed but no endpoints assigned
+ # * Description for a tag added at multiple locations => Only one description displayed.
+ list_of_ctrl = set()
for endpoints in ENDPOINT_MAP.values():
for endpoint in endpoints:
if endpoint.is_api or all_endpoints:
- ctrl_names.add(endpoint.group)
+ list_of_ctrl.add(endpoint.ctrl)
+
+ TAG_MAP = {}
+ for ctrl in list_of_ctrl:
+ tag_name = ctrl.__name__
+ tag_descr = ""
+ if hasattr(ctrl, 'doc_info'):
+ if ctrl.doc_info['tag']:
+ tag_name = ctrl.doc_info['tag']
+ tag_descr = ctrl.doc_info['tag_descr']
+ if tag_name not in TAG_MAP or not TAG_MAP[tag_name]:
+ TAG_MAP[tag_name] = tag_descr
+
+ tags = [{'name': k, 'description': v if v else "*No description available*"}
+ for k, v in TAG_MAP.items()]
+ tags.sort(key=lambda e: e['name'])
+ return tags
- return [{'name': name, 'description': ""}
- for name in sorted(ctrl_names)]
+ @classmethod
+ def _get_tag(cls, endpoint):
+ """ Returns the name of a tag to assign to a path. """
+ ctrl = endpoint.ctrl
+ func = endpoint.func
+ tag = ctrl.__name__
+ if hasattr(func, 'doc_info') and func.doc_info['tag']:
+ tag = func.doc_info['tag']
+ elif hasattr(ctrl, 'doc_info') and ctrl.doc_info['tag']:
+ tag = ctrl.doc_info['tag']
+ return tag
@classmethod
def _gen_type(cls, param):
"""
Generates the type of parameter based on its name and default value,
using very simple heuristics.
+ Used if type is not explicitly defined.
"""
param_name = param['name']
def_value = param['default'] if 'default' in param else None
return "string"
@classmethod
- def _gen_body_param(cls, body_params):
- required = [p['name'] for p in body_params if p['required']]
+ # isinstance doesn't work: input is always <type 'type'>.
+ def _type_to_str(cls, type_as_type):
+ """ Used if type is explcitly defined. """
+ if type_as_type is str:
+ type_as_str = 'string'
+ elif type_as_type is int:
+ type_as_str = 'integer'
+ elif type_as_type is bool:
+ type_as_str = 'boolean'
+ elif type_as_type is list or type_as_type is tuple:
+ type_as_str = 'array'
+ elif type_as_type is float:
+ type_as_str = 'number'
+ else:
+ type_as_str = 'object'
+ return type_as_str
- props = {}
- for p in body_params:
- props[p['name']] = {
- 'type': cls._gen_type(p)
- }
- if 'default' in p:
- props[p['name']]['default'] = p['default']
+ @classmethod
+ def _add_param_info(cls, parameters, p_info):
+ # Cases to consider:
+ # * Parameter name (if not nested) misspelt in decorator => parameter not displayed
+ # * Sometime a parameter is used for several endpoints (e.g. fs_id in CephFS).
+ # Currently, there is no possiblity of reuse. Should there be?
+ # But what if there are two parameters with same name but different functionality?
+ """
+ Adds explicitly desrcibed information for parameters of an endpoint.
+
+ There are two cases:
+ * Either the parameter in p_info corresponds to an endpoint parameter. Implicit information
+ has higher priority, so only information that doesn't already exist is added.
+ * Or the parameter in p_info describes a nested parameter inside an endpoint parameter.
+ In that case there is no implcit information at all so all explicitly described info needs
+ to be added.
+ """
+ for p in p_info:
+ if not p['nested']:
+ for parameter in parameters:
+ if p['name'] == parameter['name']:
+ parameter['type'] = p['type']
+ parameter['description'] = p['description']
+ if 'nested_params' in p:
+ parameter['nested_params'] = cls._add_param_info([], p['nested_params'])
+ else:
+ nested_p = {
+ 'name': p['name'],
+ 'type': p['type'],
+ 'description': p['description'],
+ 'required': p['required'],
+ }
+ if 'default' in p:
+ nested_p['default'] = p['default']
+ if 'nested_params' in p:
+ nested_p['nested_params'] = cls._add_param_info([], p['nested_params'])
+ parameters.append(nested_p)
- if not props:
- return None
+ return parameters
- return {
- 'title': '',
- 'type': "object",
- 'required': required,
- 'properties': props
+ @classmethod
+ def _gen_schema_for_content(cls, params):
+ """
+ Generates information to the content-object in OpenAPI Spec.
+ Used to for request body and responses.
+ """
+ required_params = []
+ properties = {}
+
+ for param in params:
+ if param['required']:
+ required_params.append(param['name'])
+
+ props = {}
+ if 'type' in param:
+ props['type'] = cls._type_to_str(param['type'])
+ if 'nested_params' in param:
+ if props['type'] == 'array': # dict in array
+ props['items'] = cls._gen_schema_for_content(param['nested_params'])
+ else: # dict in dict
+ props = cls._gen_schema_for_content(param['nested_params'])
+ elif props['type'] == 'object': # e.g. [int]
+ props['type'] = 'array'
+ props['items'] = {'type': cls._type_to_str(param['type'][0])}
+ else:
+ props['type'] = cls._gen_type(param)
+ if 'description' in param:
+ props['description'] = param['description']
+ if 'default' in param:
+ props['default'] = param['default']
+ properties[param['name']] = props
+
+ schema = {
+ 'type': 'object',
+ 'properties': properties,
}
+ if required_params:
+ schema['required'] = required_params
+ return schema
@classmethod
- def _gen_responses_descriptions(cls, method):
+ def _gen_responses(cls, method, resp_object=None):
resp = {
'400': {
"description": "Operation exception. Please check the "
resp['202'] = {'description': "Operation is still executing."
" Please check the task queue."}
+ if resp_object:
+ for status_code, response_body in resp_object.items():
+ resp[status_code].update({
+ 'content': {
+ 'application/json': {
+ 'schema': cls._gen_schema_for_content(response_body)}}})
+
return resp
@classmethod
- def _gen_param(cls, param, ptype):
- res = {
- 'name': param['name'],
- 'in': ptype,
- 'schema': {
- 'type': cls._gen_type(param)
+ def _gen_params(cls, params, location):
+ parameters = []
+ for param in params:
+ if 'type' in param:
+ _type = cls._type_to_str(param['type'])
+ else:
+ _type = cls._gen_type(param)
+ if 'description' in param:
+ descr = param['description']
+ else:
+ descr = "*No description available*"
+ res = {
+ 'name': param['name'],
+ 'in': location,
+ 'schema': {
+ 'type': _type
+ },
+ 'description': descr
}
- }
- if param['required']:
- res['required'] = True
- elif param['default'] is None:
- res['allowEmptyValue'] = True
- else:
- res['default'] = param['default']
- return res
+ if param['required']:
+ res['required'] = True
+ elif param['default'] is None:
+ res['allowEmptyValue'] = True
+ else:
+ res['default'] = param['default']
+ parameters.append(res)
- def _gen_spec(self, all_endpoints=False, baseUrl=""):
- if all_endpoints:
- baseUrl = ""
- METHOD_ORDER = ['get', 'post', 'put', 'delete']
- host = cherrypy.request.base
- host = host[host.index(':')+3:]
- logger.debug("DOCS: Host: %s", host)
+ return parameters
+ @classmethod
+ def _gen_paths(cls, all_endpoints, baseUrl):
+ METHOD_ORDER = ['get', 'post', 'put', 'delete']
paths = {}
for path, endpoints in sorted(list(ENDPOINT_MAP.items()),
key=lambda p: p[0]):
break
method = endpoint.method
+ func = endpoint.func
+
+ summary = "No description available"
+ resp = {}
+ p_info = []
+ if hasattr(func, 'doc_info'):
+ if func.doc_info['summary']:
+ summary = func.doc_info['summary']
+ resp = func.doc_info['response']
+ p_info = func.doc_info['parameters']
params = []
- params.extend([self._gen_param(p, 'path')
- for p in endpoint.path_params])
- params.extend([self._gen_param(p, 'query')
- for p in endpoint.query_params])
+ if endpoint.path_params:
+ params.extend(
+ cls._gen_params(
+ cls._add_param_info(endpoint.path_params, p_info), 'path'))
+ if endpoint.query_params:
+ params.extend(
+ cls._gen_params(
+ cls._add_param_info(endpoint.query_params, p_info), 'query'))
methods[method.lower()] = {
- 'tags': [endpoint.group],
- 'summary': "",
- 'consumes': [
- "application/json"
- ],
- 'produces': [
- "application/json"
- ],
+ 'tags': [cls._get_tag(endpoint)],
+ 'summary': summary,
+ 'description': func.__doc__,
'parameters': params,
- 'responses': self._gen_responses_descriptions(method)
+ 'responses': cls._gen_responses(method, resp)
}
if method.lower() in ['post', 'put']:
- body_params = self._gen_body_param(endpoint.body_params)
- if body_params:
+ if endpoint.body_params:
+ body_params = cls._add_param_info(endpoint.body_params, p_info)
methods[method.lower()]['requestBody'] = {
'content': {
'application/json': {
- 'schema': body_params
- }
- }
- }
+ 'schema': cls._gen_schema_for_content(body_params)}}}
if endpoint.is_secure:
methods[method.lower()]['security'] = [{'jwt': []}]
if not skip:
paths[path[len(baseUrl):]] = methods
+ return paths
+
+ def _gen_spec(self, all_endpoints=False, baseUrl=""):
+ if all_endpoints:
+ baseUrl = ""
+
+ host = cherrypy.request.base
+ host = host[host.index(':')+3:]
+ logger.debug("DOCS: Host: %s", host)
+
+ paths = self._gen_paths(all_endpoints, baseUrl)
+
if not baseUrl:
baseUrl = "/"
scheme = 'https'
- ssl = strtobool(mgr.get_localized_module_option('ssl', 'True'))
+ ssl = mgr.get_localized_module_option('ssl', 'True')
if not ssl:
scheme = 'http'
{{
box-sizing: inherit;
}}
-
body {{
margin:0;
background: #fafafa;
</style>
</head>
<body>
-
<div id="swagger-ui"></div>
-
<script src="//unpkg.com/swagger-ui-dist@3/swagger-ui-bundle.js">
</script>
<script>
--- /dev/null
+# # -*- coding: utf-8 -*-
+from __future__ import absolute_import
+
+from .helper import ControllerTestCase
+from ..controllers import RESTController, ApiController, Endpoint, EndpointDoc, ControllerDoc
+from ..controllers.docs import Docs
+
+
+# Dummy controller and endpoint that can be assigned with @EndpointDoc and @GroupDoc
+@ControllerDoc("Group description", group="FooGroup")
+@ApiController("/doctest/", secure=False)
+class DecoratedController(RESTController):
+ @EndpointDoc(
+ description="Endpoint description",
+ group="BarGroup",
+ parameters={
+ 'parameter': (int, "Description of parameter"),
+ },
+ responses={
+ 200: {
+ 'resp': (str, 'Description of response')
+ },
+ },
+ )
+ @Endpoint(json_response=False)
+ def decorated_func(self, parameter):
+ pass
+
+
+# To assure functionality of @EndpointDoc, @GroupDoc
+class DocDecoratorsTest(ControllerTestCase):
+ @classmethod
+ def setup_server(cls):
+ cls.setup_controllers([DecoratedController, Docs], "/test")
+
+ def test_group_info_attr(self):
+ testCtrl = DecoratedController()
+ self.assertTrue(hasattr(testCtrl, 'doc_info'))
+ self.assertIn('tag_descr', testCtrl.doc_info)
+ self.assertIn('tag', testCtrl.doc_info)
+
+ def test_endpoint_info_attr(self):
+ testCtrl = DecoratedController()
+ testEndpoint = testCtrl.decorated_func
+ self.assertTrue(hasattr(testEndpoint, 'doc_info'))
+ self.assertIn('summary', testEndpoint.doc_info)
+ self.assertIn('tag', testEndpoint.doc_info)
+ self.assertIn('parameters', testEndpoint.doc_info)
+ self.assertIn('response', testEndpoint.doc_info)
+
+
+# To assure functionality of Docs.py
+# pylint: disable=protected-access
+class DocsTest(ControllerTestCase):
+ @classmethod
+ def setup_server(cls):
+ cls.setup_controllers([Docs], "/test")
+
+ def test_type_to_str(self):
+ self.assertEqual(Docs()._type_to_str(str), "string")
+
+ def test_gen_paths(self):
+ outcome = Docs()._gen_paths(False, "")['/api/doctest//decorated_func/{parameter}']['get']
+ self.assertIn('tags', outcome)
+ self.assertIn('summary', outcome)
+ self.assertIn('parameters', outcome)
+ self.assertIn('responses', outcome)
+
+ def test_gen_tags(self):
+ outcome = Docs()._gen_tags(False)[0]
+ self.assertEqual({'description': 'Group description', 'name': 'FooGroup'}, outcome)
projects / ceph.git / commitdiff