diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
commit | 19fcec84d8d7d21e796c7624e521b60d28ee21ed (patch) | |
tree | 42d26aa27d1e3f7c0b8bd3fd14e7d7082f5008dc /src/pybind/mgr/dashboard/controllers/_helpers.py | |
parent | Initial commit. (diff) | |
download | ceph-upstream/16.2.11+ds.tar.xz ceph-upstream/16.2.11+ds.zip |
Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | src/pybind/mgr/dashboard/controllers/_helpers.py | 127 |
1 files changed, 127 insertions, 0 deletions
diff --git a/src/pybind/mgr/dashboard/controllers/_helpers.py b/src/pybind/mgr/dashboard/controllers/_helpers.py new file mode 100644 index 000000000..5ec49ee97 --- /dev/null +++ b/src/pybind/mgr/dashboard/controllers/_helpers.py @@ -0,0 +1,127 @@ +import collections +import json +import logging +import re +from functools import wraps + +import cherrypy +from ceph_argparse import ArgumentFormat # pylint: disable=import-error + +from ..exceptions import DashboardException +from ..tools import getargspec + +logger = logging.getLogger(__name__) + + +ENDPOINT_MAP = collections.defaultdict(list) # type: dict + + +def _get_function_params(func): + """ + Retrieves the list of parameters declared in function. + Each parameter is represented as dict with keys: + * name (str): the name of the parameter + * required (bool): whether the parameter is required or not + * default (obj): the parameter's default value + """ + fspec = getargspec(func) + + func_params = [] + nd = len(fspec.args) if not fspec.defaults else -len(fspec.defaults) + for param in fspec.args[1:nd]: + func_params.append({'name': param, 'required': True}) + + if fspec.defaults: + for param, val in zip(fspec.args[nd:], fspec.defaults): + func_params.append({ + 'name': param, + 'required': False, + 'default': val + }) + + return func_params + + +def generate_controller_routes(endpoint, mapper, base_url): + inst = endpoint.inst + ctrl_class = endpoint.ctrl + + if endpoint.proxy: + conditions = None + else: + conditions = dict(method=[endpoint.method]) + + # base_url can be empty or a URL path that starts with "/" + # we will remove the trailing "/" if exists to help with the + # concatenation with the endpoint url below + if base_url.endswith("/"): + base_url = base_url[:-1] + + endp_url = endpoint.url + + if endp_url.find("/", 1) == -1: + parent_url = "{}{}".format(base_url, endp_url) + else: + parent_url = "{}{}".format(base_url, endp_url[:endp_url.find("/", 1)]) + + # parent_url might be of the form "/.../{...}" where "{...}" is a path parameter + # we need to remove the path parameter definition + parent_url = re.sub(r'(?:/\{[^}]+\})$', '', parent_url) + if not parent_url: # root path case + parent_url = "/" + + url = "{}{}".format(base_url, endp_url) + + logger.debug("Mapped [%s] to %s:%s restricted to %s", + url, ctrl_class.__name__, endpoint.action, + endpoint.method) + + ENDPOINT_MAP[endpoint.url].append(endpoint) + + name = ctrl_class.__name__ + ":" + endpoint.action + mapper.connect(name, url, controller=inst, action=endpoint.action, + conditions=conditions) + + # adding route with trailing slash + name += "/" + url += "/" + mapper.connect(name, url, controller=inst, action=endpoint.action, + conditions=conditions) + + return parent_url + + +def json_error_page(status, message, traceback, version): + cherrypy.response.headers['Content-Type'] = 'application/json' + return json.dumps(dict(status=status, detail=message, traceback=traceback, + version=version)) + + +def allow_empty_body(func): # noqa: N802 + """ + The POST/PUT request methods decorated with ``@allow_empty_body`` + are allowed to send empty request body. + """ + # pylint: disable=protected-access + try: + func._cp_config['tools.json_in.force'] = False + except (AttributeError, KeyError): + func._cp_config = {'tools.json_in.force': False} + return func + + +def validate_ceph_type(validations, component=''): + def decorator(func): + @wraps(func) + def validate_args(*args, **kwargs): + input_values = kwargs + for key, ceph_type in validations: + try: + ceph_type.valid(input_values[key]) + except ArgumentFormat as e: + raise DashboardException(msg=e, + code='ceph_type_not_valid', + component=component) + return func(*args, **kwargs) + return validate_args + return decorator |