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
|
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
|
