# Copyright 2018 Objectif Libre
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import importlib
import itertools
import flask
import flask_restful
import voluptuous
from werkzeug import exceptions
from cloudkitty.api import v2 as v2_api
from cloudkitty.utils import json
[docs]
class SingleQueryParam(object):
"""Voluptuous validator allowing to validate unique query parameters.
This validator checks that a URL query parameter is provided only once,
verifies its type and returns it directly, instead of returning a list
containing a single element.
Note that this validator uses ``voluptuous.Coerce`` internally and thus
should not be used together with
``cloudkitty.utils.validation.get_string_type`` in python2.
:param param_type: Type of the query parameter
"""
def __init__(self, param_type):
self._validate = voluptuous.Coerce(param_type)
def __call__(self, v):
if not isinstance(v, list):
v = [v]
if len(v) != 1:
raise voluptuous.LengthInvalid('length of value must be 1')
output = v[0]
return self._validate(output)
class MultiQueryParam(object):
"""Voluptuous validator allowing to validate multiple query parameters.
This validator splits comma-separated query parameter into lists,
verifies their type and returns it directly, instead of returning a list
containing a single element.
Note that this validator uses ``voluptuous.Coerce`` internally and thus
should not be used together with
``cloudkitty.utils.validation.get_string_type`` in python2.
:param param_type: Type of the query parameter
"""
def __init__(self, param_type):
self._validate = lambda x: list(map(voluptuous.Coerce(param_type), x))
def __call__(self, v):
if not isinstance(v, list):
v = [v]
output = itertools.chain(*[elem.split(',') for elem in v])
return self._validate(output)
class DictQueryParam(object):
"""Voluptuous helper to validate dict query params.
This validator converts a dict query parameter to a python dict.
:param key_type: Type of the dict keys
:param val_type: Type of the dict values
:param unique_values: Defaults to True. Set to True if each key should
contain only one value
:type unique_values: bool
"""
def __init__(self, key_type, val_type, unique_values=True):
self._kval = voluptuous.Coerce(key_type)
self._unique_val = unique_values
if self._unique_val:
self._vval = voluptuous.Coerce(val_type)
else:
def __vval(values):
return [voluptuous.Coerce(val_type)(v) for v in values]
self._vval = __vval
@staticmethod
def _append(output, key, val):
if key in output.keys():
output[key].append(val)
else:
output[key] = [val]
return output
def __call__(self, v):
if not isinstance(v, list):
v = [v]
tokens = itertools.chain(*[elem.split(',') for elem in v])
output = {}
for token in tokens:
try:
key, val = token.split(':')
except ValueError: # Not enough or too many values to unpack
raise voluptuous.DictInvalid(
'invalid key:value association {}'.format(token))
if key in output.keys():
if self._unique_val:
raise voluptuous.DictInvalid(
'key {} already provided'.format(key))
if self._unique_val:
output[key] = val
else:
output = self._append(output, key, val)
return {self._kval(k): self._vval(v) for k, v in output.items()}
class SingleDictQueryParam(DictQueryParam):
def __init__(self, key_type, val_type):
super(SingleDictQueryParam, self).__init__(key_type=key_type,
val_type=val_type,
unique_values=True)
class MultiDictQueryParam(DictQueryParam):
def __init__(self, key_type, val_type):
super(MultiDictQueryParam, self).__init__(key_type=key_type,
val_type=val_type,
unique_values=False)
[docs]
def paginated(func):
"""Helper function for pagination.
Adds two parameters to the decorated function:
* ``offset``: int >=0. Defaults to 0.
* ``limit``: int >=1. Defaults to 100.
Example usage::
class Example(base.BaseResource):
@api_utils.paginated
@api_utils.add_output_schema({
voluptuous.Required(
'message',
default='This is an example endpoint',
): validation_utils.get_string_type(),
})
def get(self, offset=0, limit=100):
# [...]
"""
return add_input_schema('query', {
voluptuous.Required('offset', default=0): voluptuous.All(
SingleQueryParam(int), voluptuous.Range(min=0)),
voluptuous.Required('limit', default=100): voluptuous.All(
SingleQueryParam(int), voluptuous.Range(min=1)),
})(func)
[docs]
def add_output_schema(schema):
"""Add a voluptuous schema validation on a method's output
Example usage::
class Example(base.BaseResource):
@api_utils.add_output_schema({
voluptuous.Required(
'message',
default='This is an example endpoint',
): validation_utils.get_string_type(),
})
def get(self):
return {}
:param schema: Schema to apply to the method's output
:type schema: dict
"""
schema = voluptuous.Schema(schema)
def decorator(f):
def wrap(*args, **kwargs):
resp = f(*args, **kwargs)
return schema(resp)
return wrap
return decorator
class ResourceNotFound(Exception):
"""Exception raised when a resource is not found"""
def __init__(self, module, resource_class):
msg = 'Resource {r} was not found in module {m}'.format(
r=resource_class,
m=module,
)
super(ResourceNotFound, self).__init__(msg)
def _load_resource(module, resource_class):
try:
module = importlib.import_module(module)
except ImportError:
raise ResourceNotFound(module, resource_class)
resource = getattr(module, resource_class, None)
if resource is None:
raise ResourceNotFound(module, resource_class)
return resource
def output_json(data, code, headers=None):
"""Helper function for api endpoint json serialization"""
resp = flask.make_response(json.dumps(data), code)
resp.headers.extend(headers or {})
return resp
def _get_blueprint_and_api(module_name):
endpoint_name = module_name.split('.')[-1]
blueprint = flask.Blueprint(endpoint_name, module_name)
api = flask_restful.Api(blueprint)
# Using cloudkitty.json instead of json for serialization
api.representation('application/json')(output_json)
return blueprint, api
[docs]
def do_init(app, blueprint_name, resources):
"""Registers a new Blueprint containing one or several resources to app.
:param app: Flask app in which the Blueprint should be registered
:type app: flask.Flask
:param blueprint_name: Name of the blueprint to create
:type blueprint_name: str
:param resources: Resources to add to the Blueprint's Api
:type resources: list of dicts matching
``cloudkitty.api.v2.RESOURCE_SCHEMA``
"""
blueprint, api = _get_blueprint_and_api(blueprint_name)
schema = voluptuous.Schema([v2_api.RESOURCE_SCHEMA])
for resource_info in schema(resources):
resource = _load_resource(resource_info['module'],
resource_info['resource_class'])
if resource_info['url'] and not resource_info['url'].startswith('/'):
resource_info['url'] = '/' + resource_info['url']
api.add_resource(resource, resource_info['url'])
if not blueprint_name.startswith('/'):
blueprint_name = '/' + blueprint_name
app.register_blueprint(blueprint, url_prefix=blueprint_name)
