jsexpose() - Code Metrics - Inspection of "Merge pull request #2871 from LindsayHill/typo_fix" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2ca50f...3fcc25 )

by Plexxi

created 2016-08-17 06:42 UTC

jsexpose() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

106

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	19
c	1
b	0
f	0
dl	0
loc	106
rs	2

How to fix Long Method Complexity

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

        # Note: We use type() instead of self.__class__ since self.__class__ confuses pylint
        return type(self)(**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',
             method=None):
    """
    :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 body_cls:
                if pecan.request.body:
                    data = pecan.request.json

                    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)
                else:
                    obj = None

                more.append(obj)

            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)

            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			# Note: We use type() instead of self.__class__ since self.__class__ confuses pylint
87			return type(self)(**cleaned)
88
89			@classmethod
90			def _from_model(cls, model, mask_secrets=False):
91			doc = util_mongodb.unescape_chars(model.to_mongo())
92
93			if '_id' in doc:
94			doc['id'] = str(doc.pop('_id'))
95
96			if mask_secrets and cfg.CONF.log.mask_secrets:
97			doc = model.mask_secrets(value=doc)
98
99			return doc
100
101			@classmethod
102			def from_model(cls, model, mask_secrets=False):
103			"""
104			Create API model class instance for the provided DB model instance.
105
106			:param model: DB model class instance.
107			:type model: :class:`StormFoundationDB`
108
109			:param mask_secrets: True to mask secrets in the resulting instance.
110			:type mask_secrets: ``boolean``
111			"""
112			doc = cls._from_model(model=model, mask_secrets=mask_secrets)
113			attrs = {attr: value for attr, value in six.iteritems(doc) if value is not None}
114
115			return cls(**attrs)
116
117			@classmethod
118			def to_model(cls, doc):
119			"""
120			Create a model class instance for the provided MongoDB document.
121
122			:param doc: MongoDB document.
123			"""
124			raise NotImplementedError()
125
126
127			class APIUIDMixin(object):
128			""""
129			Mixin class for retrieving UID for API objects.
130			"""
131
132			def get_uid(self):
133			# TODO: This is not the most efficient approach - refactor this functionality into util
134			# module and re-use it here and in the DB model
135			resource_db = self.to_model(self)
136			resource_uid = resource_db.get_uid()
137			return resource_uid
138
139			def get_pack_uid(self):
140			# TODO: This is not the most efficient approach - refactor this functionality into util
141			# module and re-use it here and in the DB model
142			resource_db = self.to_model(self)
143			pack_uid = resource_db.get_pack_uid()
144			return pack_uid
145
146
147			def cast_argument_value(value_type, value):
148			if value_type == bool:
149			def cast_func(value):
150			value = str(value)
151			return value.lower() in ['1', 'true']
152			else:
153			cast_func = value_type
154
155			result = cast_func(value)
156			return result
157
158
159			def get_controller_args_for_types(func, arg_types, args, kwargs):
160			"""
161			Build a list of arguments and dictionary of keyword arguments which are passed to the
162			controller method based on the arg_types specification.
163
164			Note: args argument is mutated in place.
165			"""
166			result_args = []
167			result_kwargs = {}
168
169			argspec = inspect.getargspec(func)
170			names = argspec.args[1:] # Note: we skip "self"
171
172			for index, name in enumerate(names):
173			# 1. Try kwargs first
174			if name in kwargs:
175			try:
176			value = kwargs[name]
177			value_type = arg_types[index]
178			value = cast_argument_value(value_type=value_type, value=value)
179			result_kwargs[name] = value
180			except IndexError:
181			LOG.warning("Type definition for '%s' argument of '%s' is missing.",
182			name, func.__name__)
183
184			continue
185
186			# 2. Try positional args
187			try:
188			value = args.pop(0)
189			value_type = arg_types[index]
190			value = cast_argument_value(value_type=value_type, value=value)
191			result_args.append(value)
192			except IndexError:
193			LOG.warning("Type definition for '%s' argument of '%s' is missing.",
194			name, func.__name__)
195
196			return result_args, result_kwargs
197
198
199			def jsexpose(arg_types=None, body_cls=None, status_code=None, content_type='application/json',
200			method=None):
201			"""
202			:param arg_types: A list of types for the function arguments (e.g. [str, str, int, bool]).
203			:type arg_types: ``list``
204
205			:param body_cls: Request body class. If provided, this class will be used to create an instance
206			out of the request body.
207			:type body_cls: :class:`object`
208
209			:param status_code: Response status code.
210			:type status_code: ``int``
211
212			:param content_type: Response content type.
213			:type content_type: ``str``
214			"""
215			pecan_json_decorate = pecan.expose(
216			content_type=content_type,
217			generic=False)
218
219			def decorate(f):
220			@functools.wraps(f)
221			def callfunction(args, *kwargs):
222			function_name = f.__name__
223			args = list(args)
224			more = [args.pop(0)]
225
226			def cast_value(value_type, value):
227			if value_type == bool:
228			def cast_func(value):
229			return value.lower() in ['1', 'true']
230			else:
231			cast_func = value_type
232
233			result = cast_func(value)
234			return result
235
236			if body_cls:
237			if pecan.request.body:
238			data = pecan.request.json
239
240			obj = body_cls(**data)
241			try:
242			obj = obj.validate()
243			except (jsonschema.ValidationError, ValueError) as e:
244			raise exc.HTTPBadRequest(detail=e.message,
245			comment=traceback.format_exc())
246			except Exception as e:
247			raise exc.HTTPInternalServerError(detail=e.message,
248			comment=traceback.format_exc())
249
250			# Set default pack if one is not provided for resource create
251			if function_name == 'post' and not hasattr(obj, 'pack'):
252			extra = {
253			'resource_api': obj,
254			'default_pack_name': DEFAULT_PACK_NAME
255			}
256			LOG.debug('Pack not provided in the body, setting a default pack name',
257			extra=extra)
258			setattr(obj, 'pack', DEFAULT_PACK_NAME)
259			else:
260			obj = None
261
262			more.append(obj)
263
264			if arg_types:
265			# Cast and transform arguments based on the provided arg_types specification
266			result_args, result_kwargs = get_controller_args_for_types(func=f,
267			arg_types=arg_types,
268			args=args,
269			kwargs=kwargs)
270			more = more + result_args
271			kwargs.update(result_kwargs)
272
273			args = tuple(more) + tuple(args)
274
275			noop_codes = [http_client.NOT_IMPLEMENTED,
276			http_client.METHOD_NOT_ALLOWED,
277			http_client.FORBIDDEN]
278
279			if status_code and status_code in noop_codes:
280			pecan.response.status = status_code
281			return json_encode(None)
282
283			try:
284			result = f(args, *kwargs)
285			except TypeError as e:
286			e = get_exception_for_type_error(func=f, exc=e)
287			raise e
288
289			if status_code:
290			pecan.response.status = status_code
291			if content_type == 'application/json':
292			if is_debugging_enabled():
293			indent = 4
294			else:
295			indent = None
296			return json_encode(result, indent=indent)
297			else:
298			return result
299
300			pecan_json_decorate(callfunction)
301
302			return callfunction
303
304			return decorate
305

StackStorm / st2

Push — master ( 2ca50f...3fcc25 )

jsexpose() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like