cast_func() - Code Metrics - Inspection of "Merge pull request #2659 from StackStorm/jsexpose_..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 82f97f...71d0f8 )

by Plexxi

created 2016-05-03 06:42 UTC

cast_func() A

↳ Parent: cast_argument_value()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	1
c	2
b	0
f	0
dl	0
loc	3
rs	10

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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 abc
import functools
import inspect

import jsonschema
import six
from six.moves import http_client
from webob import exc
import pecan
import traceback
from oslo_config import cfg

from st2common.constants.pack import DEFAULT_PACK_NAME
from st2common.util import mongoescape as util_mongodb
from st2common.util import schema as util_schema
from st2common.util.debugging import is_enabled as is_debugging_enabled
from st2common.util.jsonify import json_encode
from st2common.util.api import get_exception_for_type_error
from st2common import log as logging

__all__ = [
    'BaseAPI',

    'APIUIDMixin',

    'jsexpose'
]


LOG = logging.getLogger(__name__)


@six.add_metaclass(abc.ABCMeta)
class BaseAPI(object):
    schema = abc.abstractproperty

    def __init__(self, **kw):
        for key, value in kw.items():
            setattr(self, key, value)

    def __repr__(self):
        name = type(self).__name__
        attrs = ', '.join("'%s': %r" % item for item in six.iteritems(vars(self)))
        # The format here is so that eval can be applied.
        return "%s(**{%s})" % (name, attrs)

    def __str__(self):
        name = type(self).__name__
        attrs = ', '.join("%s=%r" % item for item in six.iteritems(vars(self)))

        return "%s[%s]" % (name, attrs)

    def __json__(self):
        return vars(self)

    def validate(self):
        """
        Perform validation and return cleaned object on success.

        Note: This method doesn't mutate this object in place, but it returns a new one.

        :return: Cleaned / validated object.
        """
        schema = getattr(self, 'schema', {})
        attributes = vars(self)

        cleaned = util_schema.validate(instance=attributes, schema=schema,
                                       cls=util_schema.CustomValidator, use_default=True,
                                       allow_default_none=True)

        return self.__class__(**cleaned)

    @classmethod
    def _from_model(cls, model, mask_secrets=False):
        doc = util_mongodb.unescape_chars(model.to_mongo())

        if '_id' in doc:
            doc['id'] = str(doc.pop('_id'))

        if mask_secrets and cfg.CONF.log.mask_secrets:
            doc = model.mask_secrets(value=doc)

        return doc

    @classmethod
    def from_model(cls, model, mask_secrets=False):
        """
        Create API model class instance for the provided DB model instance.

        :param model: DB model class instance.
        :type model: :class:`StormFoundationDB`

        :param mask_secrets: True to mask secrets in the resulting instance.
        :type mask_secrets: ``boolean``
        """
        doc = cls._from_model(model=model, mask_secrets=mask_secrets)
        attrs = {attr: value for attr, value in six.iteritems(doc) if value is not None}

        return cls(**attrs)

    @classmethod
    def to_model(cls, doc):
        """
        Create a model class instance for the provided MongoDB document.

        :param doc: MongoDB document.
        """
        raise NotImplementedError()


class APIUIDMixin(object):
    """"
    Mixin class for retrieving UID for API objects.
    """

    def get_uid(self):
        # TODO: This is not the most efficient approach - refactor this functionality into util
        # module and re-use it here and in the DB model
        resource_db = self.to_model(self)
        resource_uid = resource_db.get_uid()
        return resource_uid

    def get_pack_uid(self):
        # TODO: This is not the most efficient approach - refactor this functionality into util
        # module and re-use it here and in the DB model
        resource_db = self.to_model(self)
        pack_uid = resource_db.get_pack_uid()
        return pack_uid


def cast_argument_value(value_type, value):
    if value_type == bool:
        def cast_func(value):
            value = str(value)
            return value.lower() in ['1', 'true']
    else:
        cast_func = value_type

    result = cast_func(value)
    return result


def get_controller_args_for_types(func, arg_types, args, kwargs):
    """
    Build a list of arguments and dictionary of keyword arguments which are passed to the
    controller method based on the arg_types specification.

    Note: args argument is mutated in place.
    """
    result_args = []
    result_kwargs = {}

    argspec = inspect.getargspec(func)
    names = argspec.args[1:]  # Note: we skip "self"

    for index, name in enumerate(names):
        # 1. Try kwargs first
        if name in kwargs:
            try:
                value = kwargs[name]
                value_type = arg_types[index]
                value = cast_argument_value(value_type=value_type, value=value)
                result_kwargs[name] = value
            except IndexError:
                LOG.warning("Type definition for '%s' argument of '%s' is missing.",
                            name, func.__name__)

            continue

        # 2. Try positional args
        try:
            value = args.pop(0)
            value_type = arg_types[index]
            value = cast_argument_value(value_type=value_type, value=value)
            result_args.append(value)
        except IndexError:
            LOG.warning("Type definition for '%s' argument of '%s' is missing.",
                        name, func.__name__)

    return result_args, result_kwargs


def jsexpose(arg_types=None, body_cls=None, status_code=None, content_type='application/json'):
    """
    :param arg_types: A list of types for the function arguments (e.g. [str, str, int, bool]).
    :type arg_types: ``list``

    :param body_cls: Request body class. If provided, this class will be used to create an instance
                     out of the request body.
    :type body_cls: :class:`object`

    :param status_code: Response status code.
    :type status_code: ``int``

    :param content_type: Response content type.
    :type content_type: ``str``
    """
    pecan_json_decorate = pecan.expose(
        content_type=content_type,
        generic=False)

    def decorate(f):
        @functools.wraps(f)
        def callfunction(*args, **kwargs):
            function_name = f.__name__
            args = list(args)
            more = [args.pop(0)]

            def cast_value(value_type, value):
                if value_type == bool:
                    def cast_func(value):
                        return value.lower() in ['1', 'true']
                else:
                    cast_func = value_type

                result = cast_func(value)
                return result

            if arg_types:
                # Cast and transform arguments based on the provided arg_types specification
                result_args, result_kwargs = get_controller_args_for_types(func=f,
                                                                           arg_types=arg_types,
                                                                           args=args,
                                                                           kwargs=kwargs)
                more = more + result_args
                kwargs.update(result_kwargs)

            if body_cls:
                if pecan.request.body:
                    data = pecan.request.json
                else:
                    data = {}

                obj = body_cls(**data)
                try:
                    obj = obj.validate()
                except (jsonschema.ValidationError, ValueError) as e:
                    raise exc.HTTPBadRequest(detail=e.message,
                                             comment=traceback.format_exc())
                except Exception as e:
                    raise exc.HTTPInternalServerError(detail=e.message,
                                                      comment=traceback.format_exc())

                # Set default pack if one is not provided for resource create
                if function_name == 'post' and not hasattr(obj, 'pack'):
                    extra = {
                        'resource_api': obj,
                        'default_pack_name': DEFAULT_PACK_NAME
                    }
                    LOG.debug('Pack not provided in the body, setting a default pack name',
                              extra=extra)
                    setattr(obj, 'pack', DEFAULT_PACK_NAME)

                more.append(obj)

            args = tuple(more) + tuple(args)

            noop_codes = [http_client.NOT_IMPLEMENTED,
                          http_client.METHOD_NOT_ALLOWED,
                          http_client.FORBIDDEN]

            if status_code and status_code in noop_codes:
                pecan.response.status = status_code
                return json_encode(None)

            try:
                result = f(*args, **kwargs)
            except TypeError as e:
                e = get_exception_for_type_error(func=f, exc=e)
                raise e

            if status_code:
                pecan.response.status = status_code
            if content_type == 'application/json':
                if is_debugging_enabled():
                    indent = 4
                else:
                    indent = None
                return json_encode(result, indent=indent)
            else:
                return result

        pecan_json_decorate(callfunction)

        return callfunction

    return decorate


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import abc
17			import functools
18			import inspect
19
20			import jsonschema
21			import six
22			from six.moves import http_client
23			from webob import exc
24			import pecan
25			import traceback
26			from oslo_config import cfg
27
28			from st2common.constants.pack import DEFAULT_PACK_NAME
29			from st2common.util import mongoescape as util_mongodb
30			from st2common.util import schema as util_schema
31			from st2common.util.debugging import is_enabled as is_debugging_enabled
32			from st2common.util.jsonify import json_encode
33			from st2common.util.api import get_exception_for_type_error
34			from st2common import log as logging
35
36			__all__ = [
37			'BaseAPI',
38
39			'APIUIDMixin',
40
41			'jsexpose'
42			]
43
44
45			LOG = logging.getLogger(__name__)
46
47
48			@six.add_metaclass(abc.ABCMeta)
49			class BaseAPI(object):
50			schema = abc.abstractproperty
51
52			def __init__(self, **kw):
53			for key, value in kw.items():
54			setattr(self, key, value)
55
56			def __repr__(self):
57			name = type(self).__name__
58			attrs = ', '.join("'%s': %r" % item for item in six.iteritems(vars(self)))
59			# The format here is so that eval can be applied.
60			return "%s(**{%s})" % (name, attrs)
61
62			def __str__(self):
63			name = type(self).__name__
64			attrs = ', '.join("%s=%r" % item for item in six.iteritems(vars(self)))
65
66			return "%s[%s]" % (name, attrs)
67
68			def __json__(self):
69			return vars(self)
70
71			def validate(self):
72			"""
73			Perform validation and return cleaned object on success.
74
75			Note: This method doesn't mutate this object in place, but it returns a new one.
76
77			:return: Cleaned / validated object.
78			"""
79			schema = getattr(self, 'schema', {})
80			attributes = vars(self)
81
82			cleaned = util_schema.validate(instance=attributes, schema=schema,
83			cls=util_schema.CustomValidator, use_default=True,
84			allow_default_none=True)
85
86			return self.__class__(**cleaned)
87
88			@classmethod
89			def _from_model(cls, model, mask_secrets=False):
90			doc = util_mongodb.unescape_chars(model.to_mongo())
91
92			if '_id' in doc:
93			doc['id'] = str(doc.pop('_id'))
94
95			if mask_secrets and cfg.CONF.log.mask_secrets:
96			doc = model.mask_secrets(value=doc)
97
98			return doc
99
100			@classmethod
101			def from_model(cls, model, mask_secrets=False):
102			"""
103			Create API model class instance for the provided DB model instance.
104
105			:param model: DB model class instance.
106			:type model: :class:`StormFoundationDB`
107
108			:param mask_secrets: True to mask secrets in the resulting instance.
109			:type mask_secrets: ``boolean``
110			"""
111			doc = cls._from_model(model=model, mask_secrets=mask_secrets)
112			attrs = {attr: value for attr, value in six.iteritems(doc) if value is not None}
113
114			return cls(**attrs)
115
116			@classmethod
117			def to_model(cls, doc):
118			"""
119			Create a model class instance for the provided MongoDB document.
120
121			:param doc: MongoDB document.
122			"""
123			raise NotImplementedError()
124
125
126			class APIUIDMixin(object):
127			""""
128			Mixin class for retrieving UID for API objects.
129			"""
130
131			def get_uid(self):
132			# TODO: This is not the most efficient approach - refactor this functionality into util
133			# module and re-use it here and in the DB model
134			resource_db = self.to_model(self)
135			resource_uid = resource_db.get_uid()
136			return resource_uid
137
138			def get_pack_uid(self):
139			# TODO: This is not the most efficient approach - refactor this functionality into util
140			# module and re-use it here and in the DB model
141			resource_db = self.to_model(self)
142			pack_uid = resource_db.get_pack_uid()
143			return pack_uid
144
145
146			def cast_argument_value(value_type, value):
147			if value_type == bool:
148			def cast_func(value):
149			value = str(value)
150			return value.lower() in ['1', 'true']
151			else:
152			cast_func = value_type
153
154			result = cast_func(value)
155			return result
156
157
158			def get_controller_args_for_types(func, arg_types, args, kwargs):
159			"""
160			Build a list of arguments and dictionary of keyword arguments which are passed to the
161			controller method based on the arg_types specification.
162
163			Note: args argument is mutated in place.
164			"""
165			result_args = []
166			result_kwargs = {}
167
168			argspec = inspect.getargspec(func)
169			names = argspec.args[1:] # Note: we skip "self"
170
171			for index, name in enumerate(names):
172			# 1. Try kwargs first
173			if name in kwargs:
174			try:
175			value = kwargs[name]
176			value_type = arg_types[index]
177			value = cast_argument_value(value_type=value_type, value=value)
178			result_kwargs[name] = value
179			except IndexError:
180			LOG.warning("Type definition for '%s' argument of '%s' is missing.",
181			name, func.__name__)
182
183			continue
184
185			# 2. Try positional args
186			try:
187			value = args.pop(0)
188			value_type = arg_types[index]
189			value = cast_argument_value(value_type=value_type, value=value)
190			result_args.append(value)
191			except IndexError:
192			LOG.warning("Type definition for '%s' argument of '%s' is missing.",
193			name, func.__name__)
194
195			return result_args, result_kwargs
196
197
198			def jsexpose(arg_types=None, body_cls=None, status_code=None, content_type='application/json'):
199			"""
200			:param arg_types: A list of types for the function arguments (e.g. [str, str, int, bool]).
201			:type arg_types: ``list``
202
203			:param body_cls: Request body class. If provided, this class will be used to create an instance
204			out of the request body.
205			:type body_cls: :class:`object`
206
207			:param status_code: Response status code.
208			:type status_code: ``int``
209
210			:param content_type: Response content type.
211			:type content_type: ``str``
212			"""
213			pecan_json_decorate = pecan.expose(
214			content_type=content_type,
215			generic=False)
216
217			def decorate(f):
218			@functools.wraps(f)
219			def callfunction(args, *kwargs):
220			function_name = f.__name__
221			args = list(args)
222			more = [args.pop(0)]
223
224			def cast_value(value_type, value):
225			if value_type == bool:
226			def cast_func(value):
227			return value.lower() in ['1', 'true']
228			else:
229			cast_func = value_type
230
231			result = cast_func(value)
232			return result
233
234			if arg_types:
235			# Cast and transform arguments based on the provided arg_types specification
236			result_args, result_kwargs = get_controller_args_for_types(func=f,
237			arg_types=arg_types,
238			args=args,
239			kwargs=kwargs)
240			more = more + result_args
241			kwargs.update(result_kwargs)
242
243			if body_cls:
244			if pecan.request.body:
245			data = pecan.request.json
246			else:
247			data = {}
248
249			obj = body_cls(**data)
250			try:
251			obj = obj.validate()
252			except (jsonschema.ValidationError, ValueError) as e:
253			raise exc.HTTPBadRequest(detail=e.message,
254			comment=traceback.format_exc())
255			except Exception as e:
256			raise exc.HTTPInternalServerError(detail=e.message,
257			comment=traceback.format_exc())
258
259			# Set default pack if one is not provided for resource create
260			if function_name == 'post' and not hasattr(obj, 'pack'):
261			extra = {
262			'resource_api': obj,
263			'default_pack_name': DEFAULT_PACK_NAME
264			}
265			LOG.debug('Pack not provided in the body, setting a default pack name',
266			extra=extra)
267			setattr(obj, 'pack', DEFAULT_PACK_NAME)
268
269			more.append(obj)
270
271			args = tuple(more) + tuple(args)
272
273			noop_codes = [http_client.NOT_IMPLEMENTED,
274			http_client.METHOD_NOT_ALLOWED,
275			http_client.FORBIDDEN]
276
277			if status_code and status_code in noop_codes:
278			pecan.response.status = status_code
279			return json_encode(None)
280
281			try:
282			result = f(args, *kwargs)
283			except TypeError as e:
284			e = get_exception_for_type_error(func=f, exc=e)
285			raise e
286
287			if status_code:
288			pecan.response.status = status_code
289			if content_type == 'application/json':
290			if is_debugging_enabled():
291			indent = 4
292			else:
293			indent = None
294			return json_encode(result, indent=indent)
295			else:
296			return result
297
298			pecan_json_decorate(callfunction)
299
300			return callfunction
301
302			return decorate
303

StackStorm / st2

Push — develop ( 82f97f...71d0f8 )

cast_func() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like