import ceph 16.2.7

[ceph.git] / ceph / src / pybind / mgr / dashboard / controllers / _helpers.py

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