callfunction() - Code Metrics - Inspection of "Add tests for submit_debug_info.py temp dir remova..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop-v1.3.1 ( 8fa207...a1cf9b )

unknown

created 2016-05-27 15:10 UTC

callfunction() F

↳ Parent: jsexpose()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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 re
import abc
import copy
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 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 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.
    :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)
            types = copy.copy(arg_types)
            more = [args.pop(0)]

            if types:
                argspec = inspect.getargspec(f)
                names = argspec.args[1:]

                for name in names:
                    try:
                        a = args.pop(0)
                        more.append(types.pop(0)(a))
                    except IndexError:
                        try:
                            kwargs[name] = types.pop(0)(kwargs[name])
                        except IndexError:
                            LOG.warning("Type definition for '%s' argument of '%s' "
                                        "is missing.", name, f.__name__)
                        except KeyError:
                            pass

            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:
                message = str(e)
                # Invalid number of arguments passed to the function meaning invalid path was
                # requested
                # Note: The check is hacky, but it works for now.
                func_name = f.__name__
                pattern = '%s\(\) takes exactly \d+ arguments \(\d+ given\)' % (func_name)


                if re.search(pattern, message):
                    raise exc.HTTPNotFound()
                else:
                    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 re
17			import abc
18			import copy
19			import functools
20			import inspect
21
22			import jsonschema
23			import six
24			from six.moves import http_client
25			from webob import exc
26			import pecan
27			import traceback
28			from oslo_config import cfg
29
30			from st2common.constants.pack import DEFAULT_PACK_NAME
31			from st2common.util import mongoescape as util_mongodb
32			from st2common.util import schema as util_schema
33			from st2common.util.debugging import is_enabled as is_debugging_enabled
34			from st2common.util.jsonify import json_encode
35			from st2common import log as logging
36
37			__all__ = [
38			'BaseAPI',
39
40			'APIUIDMixin',
41
42			'jsexpose'
43			]
44
45
46			LOG = logging.getLogger(__name__)
47
48
49			@six.add_metaclass(abc.ABCMeta)
50			class BaseAPI(object):
51			schema = abc.abstractproperty
52
53			def __init__(self, **kw):
54			for key, value in kw.items():
55			setattr(self, key, value)
56
57			def __repr__(self):
58			name = type(self).__name__
59			attrs = ', '.join("'%s': %r" % item for item in six.iteritems(vars(self)))
60			# The format here is so that eval can be applied.
61			return "%s(**{%s})" % (name, attrs)
62
63			def __str__(self):
64			name = type(self).__name__
65			attrs = ', '.join("%s=%r" % item for item in six.iteritems(vars(self)))
66
67			return "%s[%s]" % (name, attrs)
68
69			def __json__(self):
70			return vars(self)
71
72			def validate(self):
73			"""
74			Perform validation and return cleaned object on success.
75
76			Note: This method doesn't mutate this object in place, but it returns a new one.
77
78			:return: Cleaned / validated object.
79			"""
80			schema = getattr(self, 'schema', {})
81			attributes = vars(self)
82
83			cleaned = util_schema.validate(instance=attributes, schema=schema,
84			cls=util_schema.CustomValidator, use_default=True,
85			allow_default_none=True)
86
87			return self.__class__(**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 jsexpose(arg_types=None, body_cls=None, status_code=None, content_type='application/json'):
148			"""
149			:param arg_types: A list of types for the function arguments.
150			:type arg_types: ``list``
151
152			:param body_cls: Request body class. If provided, this class will be used to create an instance
153			out of the request body.
154			:type body_cls: :class:`object`
155
156			:param status_code: Response status code.
157			:type status_code: ``int``
158
159			:param content_type: Response content type.
160			:type content_type: ``str``
161			"""
162			pecan_json_decorate = pecan.expose(
163			content_type=content_type,
164			generic=False)
165
166			def decorate(f):
167			@functools.wraps(f)
168			def callfunction(args, *kwargs):
169			function_name = f.__name__
170			args = list(args)
171			types = copy.copy(arg_types)
172			more = [args.pop(0)]
173
174			if types:
175			argspec = inspect.getargspec(f)
176			names = argspec.args[1:]
177
178			for name in names:
179			try:
180			a = args.pop(0)
181			more.append(types.pop(0)(a))
182			except IndexError:
183			try:
184			kwargs[name] = types.pop(0)(kwargs[name])
185			except IndexError:
186			LOG.warning("Type definition for '%s' argument of '%s' "
187			"is missing.", name, f.__name__)
188			except KeyError:
189			pass
190
191			if body_cls:
192			if pecan.request.body:
193			data = pecan.request.json
194			else:
195			data = {}
196
197			obj = body_cls(**data)
198			try:
199			obj = obj.validate()
200			except (jsonschema.ValidationError, ValueError) as e:
201			raise exc.HTTPBadRequest(detail=e.message,
202			comment=traceback.format_exc())
203			except Exception as e:
204			raise exc.HTTPInternalServerError(detail=e.message,
205			comment=traceback.format_exc())
206
207			# Set default pack if one is not provided for resource create
208			if function_name == 'post' and not hasattr(obj, 'pack'):
209			extra = {
210			'resource_api': obj,
211			'default_pack_name': DEFAULT_PACK_NAME
212			}
213			LOG.debug('Pack not provided in the body, setting a default pack name',
214			extra=extra)
215			setattr(obj, 'pack', DEFAULT_PACK_NAME)
216
217			more.append(obj)
218
219			args = tuple(more) + tuple(args)
220
221			noop_codes = [http_client.NOT_IMPLEMENTED,
222			http_client.METHOD_NOT_ALLOWED,
223			http_client.FORBIDDEN]
224
225			if status_code and status_code in noop_codes:
226			pecan.response.status = status_code
227			return json_encode(None)
228
229			try:
230			result = f(args, *kwargs)
231			except TypeError as e:
232			message = str(e)
233			# Invalid number of arguments passed to the function meaning invalid path was
234			# requested
235			# Note: The check is hacky, but it works for now.
236			func_name = f.__name__
237			pattern = '%s\(\) takes exactly \d+ arguments \(\d+ given\)' % (func_name)
			0 ignored issues – show Bug introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\(` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history... Bug introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\)` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history... Bug introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\d` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history...
238
239			if re.search(pattern, message):
240			raise exc.HTTPNotFound()
241			else:
242			raise e
243
244			if status_code:
245			pecan.response.status = status_code
246			if content_type == 'application/json':
247			if is_debugging_enabled():
248			indent = 4
249			else:
250			indent = None
251			return json_encode(result, indent=indent)
252			else:
253			return result
254
255			pecan_json_decorate(callfunction)
256
257			return callfunction
258
259			return decorate
260

Plexxi / st2

GitHub Access Token became invalid

Push — develop-v1.3.1 ( 8fa207...a1cf9b )

callfunction() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like