st2client.models.WebhookBranch - Code Metrics - Inspection of "Merge pull request #2357 from StackStorm/vagrant-d..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( bb5487...4fc1d0 )

by Manas

created 2016-01-06 20:30 UTC

st2client.models.WebhookBranch A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	9
Duplicated Lines	0 %

Metric	Value
dl	0
loc	9
rs	10
wmc	1

# 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 os
import json
import logging
from functools import wraps

import six

from six.moves import urllib
from st2client.utils import httpclient


LOG = logging.getLogger(__name__)


def add_auth_token_to_kwargs_from_env(func):
    @wraps(func)
    def decorate(*args, **kwargs):
        if not kwargs.get('token') and os.environ.get('ST2_AUTH_TOKEN', None):
            kwargs['token'] = os.environ.get('ST2_AUTH_TOKEN')
        return func(*args, **kwargs)
    return decorate


class Resource(object):

    # An alias to use for the resource if different than the class name.
    _alias = None

    # Display name of the resource. This may be different than its resource
    # name specifically when the resource name is composed of multiple words.
    _display_name = None

    # URL path for the resource.
    _url_path = None

    # Plural form of the resource name. This will be used to build the
    # latter part of the REST URL.
    _plural = None

    # Plural form of the resource display name.
    _plural_display_name = None

    # A list of class attributes which will be included in __repr__ return value
    _repr_attributes = []

    def __init__(self, *args, **kwargs):
        for k, v in six.iteritems(kwargs):
            setattr(self, k, v)

    def to_dict(self, exclude_attributes=None):
        """
        Return a dictionary representation of this object.

        :param exclude_attributes: Optional list of attributes to exclude.
        :type exclude_attributes: ``list``

        :rtype: ``dict``
        """
        exclude_attributes = exclude_attributes or []

        attributes = self.__dict__.keys()
        attributes = [attr for attr in attributes if not attr.startswith('__') and
                      attr not in exclude_attributes]

        result = {}
        for attribute in attributes:
            value = getattr(self, attribute, None)
            result[attribute] = value

        return result

    @classmethod
    def get_alias(cls):
        return cls._alias if cls._alias else cls.__name__

    @classmethod
    def get_display_name(cls):
        return cls._display_name if cls._display_name else cls.__name__

    @classmethod
    def get_plural_name(cls):
        if not cls._plural:
            raise Exception('The %s class is missing class attributes '
                            'in its definition.' % cls.__name__)
        return cls._plural

    @classmethod
    def get_plural_display_name(cls):
        return (cls._plural_display_name
                if cls._plural_display_name
                else cls._plural)

    @classmethod
    def get_url_path_name(cls):
        if cls._url_path:
            return cls._url_path

        return cls.get_plural_name().lower()

    def serialize(self):
        return dict((k, v)
                    for k, v in six.iteritems(self.__dict__)
                    if not k.startswith('_'))

    @classmethod
    def deserialize(cls, doc):
        if type(doc) is not dict:
            doc = json.loads(doc)
        return cls(**doc)

    def __str__(self):
        return str(self.__repr__())

    def __repr__(self):
        if not self._repr_attributes:
            return super(Resource, self).__repr__()

        attributes = []
        for attribute in self._repr_attributes:
            value = getattr(self, attribute, None)
            attributes.append('%s=%s' % (attribute, value))

        attributes = ','.join(attributes)
        class_name = self.__class__.__name__
        result = '<%s %s>' % (class_name, attributes)
        return result


class ResourceManager(object):

    def __init__(self, resource, endpoint, cacert=None, debug=False):
        self.resource = resource
        self.debug = debug
        self.client = httpclient.HTTPClient(endpoint, cacert=cacert, debug=debug)

    @staticmethod
    def handle_error(response):
        try:
            content = response.json()
            fault = content.get('faultstring', '') if content else ''
            if fault:
                response.reason += '\nMESSAGE: %s' % fault
        except Exception as e:
            response.reason += ('\nUnable to retrieve detailed message '
                                'from the HTTP response. %s\n' % str(e))
        response.raise_for_status()

    @add_auth_token_to_kwargs_from_env
    def get_all(self, **kwargs):
        # TODO: This is ugly, stop abusing kwargs
        url = '/%s' % self.resource.get_url_path_name()
        limit = kwargs.pop('limit', None)
        pack = kwargs.pop('pack', None)
        prefix = kwargs.pop('prefix', None)
        user = kwargs.pop('user', None)

        params = {}
        if limit and limit <= 0:
            limit = None
        if limit:
            params['limit'] = limit

        if pack:
            params['pack'] = pack

        if prefix:
            params['prefix'] = prefix

        if user:
            params['user'] = user

        response = self.client.get(url=url, params=params, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        return [self.resource.deserialize(item)
                for item in response.json()]

    @add_auth_token_to_kwargs_from_env
    def get_by_id(self, id, **kwargs):

        url = '/%s/%s' % (self.resource.get_url_path_name(), id)
        response = self.client.get(url, **kwargs)
        if response.status_code == 404:
            return None
        if response.status_code != 200:
            self.handle_error(response)
        return self.resource.deserialize(response.json())

    @add_auth_token_to_kwargs_from_env
    def get_property(self, id_, property_name, self_deserialize=True, **kwargs):
        """
        Gets a property of a Resource.
        id_ : Id of the resource
        property_name: Name of the property
        self_deserialize: #Implies use the deserialize method implemented by this resource.
        """
        token = None
        if kwargs:
            token = kwargs.pop('token', None)

            url = '/%s/%s/%s/?%s' % (self.resource.get_url_path_name(), id_, property_name,
                                     urllib.parse.urlencode(kwargs))
        else:
            url = '/%s/%s/%s/' % (self.resource.get_url_path_name(), id_, property_name)

        response = self.client.get(url, token=token) if token else self.client.get(url)

        if response.status_code == 404:
            return None
        if response.status_code != 200:
            self.handle_error(response)

        if self_deserialize:
            return [self.resource.deserialize(item) for item in response.json()]
        else:
            return response.json()

    @add_auth_token_to_kwargs_from_env
    def get_by_ref_or_id(self, ref_or_id, **kwargs):
        return self.get_by_id(id=ref_or_id, **kwargs)

    @add_auth_token_to_kwargs_from_env
    def query(self, **kwargs):
        if not kwargs:
            raise Exception('Query parameter is not provided.')
        if 'limit' in kwargs and kwargs.get('limit') <= 0:
            kwargs.pop('limit')
        token = kwargs.get('token', None)
        params = {}
        for k, v in six.iteritems(kwargs):
            if k != 'token':
                params[k] = v
        url = '/%s/?%s' % (self.resource.get_url_path_name(),
                           urllib.parse.urlencode(params))
        response = self.client.get(url, token=token) if token else self.client.get(url)
        if response.status_code == 404:
            return []
        if response.status_code != 200:
            self.handle_error(response)
        items = response.json()
        instances = [self.resource.deserialize(item) for item in items]
        return instances

    @add_auth_token_to_kwargs_from_env
    def get_by_name(self, name_or_id, **kwargs):
        instances = self.query(name=name_or_id, **kwargs)
        if not instances:
            return None
        else:
            if len(instances) > 1:
                raise Exception('More than one %s named "%s" are found.' %
                                (self.resource.__name__.lower(), name_or_id))
            return instances[0]

    @add_auth_token_to_kwargs_from_env
    def create(self, instance, **kwargs):
        url = '/%s' % self.resource.get_url_path_name()
        response = self.client.post(url, instance.serialize(), **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        instance = self.resource.deserialize(response.json())
        return instance

    @add_auth_token_to_kwargs_from_env
    def update(self, instance, **kwargs):
        url = '/%s/%s' % (self.resource.get_url_path_name(), instance.id)
        response = self.client.put(url, instance.serialize(), **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        instance = self.resource.deserialize(response.json())
        return instance

    @add_auth_token_to_kwargs_from_env
    def delete(self, instance, **kwargs):
        url = '/%s/%s' % (self.resource.get_url_path_name(), instance.id)
        response = self.client.delete(url, **kwargs)

        if response.status_code not in [200, 204, 404]:
            self.handle_error(response)
            return False

        return True

    @add_auth_token_to_kwargs_from_env
    def delete_by_id(self, instance_id, **kwargs):
        url = '/%s/%s' % (self.resource.get_url_path_name(), instance_id)
        response = self.client.delete(url, **kwargs)
        if response.status_code not in [200, 204, 404]:
            self.handle_error(response)
            return False
        try:
            resp_json = response.json()
            if resp_json:
                return resp_json
        except:
            pass
        return True


class ActionAliasResourceManager(ResourceManager):
    def __init__(self, resource, endpoint, cacert=None, debug=False):
class SomeParent:
    def __init__(self):
        self.x = 1

class SomeChild(SomeParent):
    def __init__(self):
        # Initialize the super class
        SomeParent.__init__(self)
        self.resource = resource
        self.debug = debug
        self.client = httpclient.HTTPClient(root=endpoint, cacert=cacert, debug=debug)


class LiveActionResourceManager(ResourceManager):
    @add_auth_token_to_kwargs_from_env
    def re_run(self, execution_id, parameters=None, **kwargs):
        url = '/%s/%s/re_run' % (self.resource.get_url_path_name(), execution_id)

        data = {}
        if parameters:
            data['parameters'] = parameters

        response = self.client.post(url, data, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)

        instance = self.resource.deserialize(response.json())
        return instance


class TriggerInstanceResourceManager(ResourceManager):
    @add_auth_token_to_kwargs_from_env
    def re_emit(self, trigger_instance_id, **kwargs):
        url = '/%s/%s/re_emit' % (self.resource.get_url_path_name(), trigger_instance_id)
        response = self.client.post(url, None)
        if response.status_code != 200:
            self.handle_error(response)
        return response.json()


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 os
17			import json
18			import logging
19			from functools import wraps
20
21			import six
22
23			from six.moves import urllib
24			from st2client.utils import httpclient
25
26
27			LOG = logging.getLogger(__name__)
28
29
30			def add_auth_token_to_kwargs_from_env(func):
31			@wraps(func)
32			def decorate(args, *kwargs):
33			if not kwargs.get('token') and os.environ.get('ST2_AUTH_TOKEN', None):
34			kwargs['token'] = os.environ.get('ST2_AUTH_TOKEN')
35			return func(args, *kwargs)
36			return decorate
37
38
39			class Resource(object):
40
41			# An alias to use for the resource if different than the class name.
42			_alias = None
43
44			# Display name of the resource. This may be different than its resource
45			# name specifically when the resource name is composed of multiple words.
46			_display_name = None
47
48			# URL path for the resource.
49			_url_path = None
50
51			# Plural form of the resource name. This will be used to build the
52			# latter part of the REST URL.
53			_plural = None
54
55			# Plural form of the resource display name.
56			_plural_display_name = None
57
58			# A list of class attributes which will be included in __repr__ return value
59			_repr_attributes = []
60
61			def __init__(self, args, *kwargs):
62			for k, v in six.iteritems(kwargs):
63			setattr(self, k, v)
64
65			def to_dict(self, exclude_attributes=None):
66			"""
67			Return a dictionary representation of this object.
68
69			:param exclude_attributes: Optional list of attributes to exclude.
70			:type exclude_attributes: ``list``
71
72			:rtype: ``dict``
73			"""
74			exclude_attributes = exclude_attributes or []
75
76			attributes = self.__dict__.keys()
77			attributes = [attr for attr in attributes if not attr.startswith('__') and
78			attr not in exclude_attributes]
79
80			result = {}
81			for attribute in attributes:
82			value = getattr(self, attribute, None)
83			result[attribute] = value
84
85			return result
86
87			@classmethod
88			def get_alias(cls):
89			return cls._alias if cls._alias else cls.__name__
90
91			@classmethod
92			def get_display_name(cls):
93			return cls._display_name if cls._display_name else cls.__name__
94
95			@classmethod
96			def get_plural_name(cls):
97			if not cls._plural:
98			raise Exception('The %s class is missing class attributes '
99			'in its definition.' % cls.__name__)
100			return cls._plural
101
102			@classmethod
103			def get_plural_display_name(cls):
104			return (cls._plural_display_name
105			if cls._plural_display_name
106			else cls._plural)
107
108			@classmethod
109			def get_url_path_name(cls):
110			if cls._url_path:
111			return cls._url_path
112
113			return cls.get_plural_name().lower()
114
115			def serialize(self):
116			return dict((k, v)
117			for k, v in six.iteritems(self.__dict__)
118			if not k.startswith('_'))
119
120			@classmethod
121			def deserialize(cls, doc):
122			if type(doc) is not dict:
123			doc = json.loads(doc)
124			return cls(**doc)
125
126			def __str__(self):
127			return str(self.__repr__())
128
129			def __repr__(self):
130			if not self._repr_attributes:
131			return super(Resource, self).__repr__()
132
133			attributes = []
134			for attribute in self._repr_attributes:
135			value = getattr(self, attribute, None)
136			attributes.append('%s=%s' % (attribute, value))
137
138			attributes = ','.join(attributes)
139			class_name = self.__class__.__name__
140			result = '<%s %s>' % (class_name, attributes)
141			return result
142
143
144			class ResourceManager(object):
145
146			def __init__(self, resource, endpoint, cacert=None, debug=False):
147			self.resource = resource
148			self.debug = debug
149			self.client = httpclient.HTTPClient(endpoint, cacert=cacert, debug=debug)
150
151			@staticmethod
152			def handle_error(response):
153			try:
154			content = response.json()
155			fault = content.get('faultstring', '') if content else ''
156			if fault:
157			response.reason += '\nMESSAGE: %s' % fault
158			except Exception as e:
159			response.reason += ('\nUnable to retrieve detailed message '
160			'from the HTTP response. %s\n' % str(e))
161			response.raise_for_status()
162
163			@add_auth_token_to_kwargs_from_env
164			def get_all(self, **kwargs):
165			# TODO: This is ugly, stop abusing kwargs
166			url = '/%s' % self.resource.get_url_path_name()
167			limit = kwargs.pop('limit', None)
168			pack = kwargs.pop('pack', None)
169			prefix = kwargs.pop('prefix', None)
170			user = kwargs.pop('user', None)
171
172			params = {}
173			if limit and limit <= 0:
174			limit = None
175			if limit:
176			params['limit'] = limit
177
178			if pack:
179			params['pack'] = pack
180
181			if prefix:
182			params['prefix'] = prefix
183
184			if user:
185			params['user'] = user
186
187			response = self.client.get(url=url, params=params, **kwargs)
188			if response.status_code != 200:
189			self.handle_error(response)
190			return [self.resource.deserialize(item)
191			for item in response.json()]
192
193			@add_auth_token_to_kwargs_from_env
194			def get_by_id(self, id, **kwargs):
			0 ignored issues – show Bug Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `id`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
195			url = '/%s/%s' % (self.resource.get_url_path_name(), id)
196			response = self.client.get(url, **kwargs)
197			if response.status_code == 404:
198			return None
199			if response.status_code != 200:
200			self.handle_error(response)
201			return self.resource.deserialize(response.json())
202
203			@add_auth_token_to_kwargs_from_env
204			def get_property(self, id_, property_name, self_deserialize=True, **kwargs):
205			"""
206			Gets a property of a Resource.
207			id_ : Id of the resource
208			property_name: Name of the property
209			self_deserialize: #Implies use the deserialize method implemented by this resource.
210			"""
211			token = None
212			if kwargs:
213			token = kwargs.pop('token', None)
214
215			url = '/%s/%s/%s/?%s' % (self.resource.get_url_path_name(), id_, property_name,
216			urllib.parse.urlencode(kwargs))
217			else:
218			url = '/%s/%s/%s/' % (self.resource.get_url_path_name(), id_, property_name)
219
220			response = self.client.get(url, token=token) if token else self.client.get(url)
221
222			if response.status_code == 404:
223			return None
224			if response.status_code != 200:
225			self.handle_error(response)
226
227			if self_deserialize:
228			return [self.resource.deserialize(item) for item in response.json()]
229			else:
230			return response.json()
231
232			@add_auth_token_to_kwargs_from_env
233			def get_by_ref_or_id(self, ref_or_id, **kwargs):
234			return self.get_by_id(id=ref_or_id, **kwargs)
235
236			@add_auth_token_to_kwargs_from_env
237			def query(self, **kwargs):
238			if not kwargs:
239			raise Exception('Query parameter is not provided.')
240			if 'limit' in kwargs and kwargs.get('limit') <= 0:
241			kwargs.pop('limit')
242			token = kwargs.get('token', None)
243			params = {}
244			for k, v in six.iteritems(kwargs):
245			if k != 'token':
246			params[k] = v
247			url = '/%s/?%s' % (self.resource.get_url_path_name(),
248			urllib.parse.urlencode(params))
249			response = self.client.get(url, token=token) if token else self.client.get(url)
250			if response.status_code == 404:
251			return []
252			if response.status_code != 200:
253			self.handle_error(response)
254			items = response.json()
255			instances = [self.resource.deserialize(item) for item in items]
256			return instances
257
258			@add_auth_token_to_kwargs_from_env
259			def get_by_name(self, name_or_id, **kwargs):
260			instances = self.query(name=name_or_id, **kwargs)
261			if not instances:
262			return None
263			else:
264			if len(instances) > 1:
265			raise Exception('More than one %s named "%s" are found.' %
266			(self.resource.__name__.lower(), name_or_id))
267			return instances[0]
268
269			@add_auth_token_to_kwargs_from_env
270			def create(self, instance, **kwargs):
271			url = '/%s' % self.resource.get_url_path_name()
272			response = self.client.post(url, instance.serialize(), **kwargs)
273			if response.status_code != 200:
274			self.handle_error(response)
275			instance = self.resource.deserialize(response.json())
276			return instance
277
278			@add_auth_token_to_kwargs_from_env
279			def update(self, instance, **kwargs):
280			url = '/%s/%s' % (self.resource.get_url_path_name(), instance.id)
281			response = self.client.put(url, instance.serialize(), **kwargs)
282			if response.status_code != 200:
283			self.handle_error(response)
284			instance = self.resource.deserialize(response.json())
285			return instance
286
287			@add_auth_token_to_kwargs_from_env
288			def delete(self, instance, **kwargs):
289			url = '/%s/%s' % (self.resource.get_url_path_name(), instance.id)
290			response = self.client.delete(url, **kwargs)
291
292			if response.status_code not in [200, 204, 404]:
293			self.handle_error(response)
294			return False
295
296			return True
297
298			@add_auth_token_to_kwargs_from_env
299			def delete_by_id(self, instance_id, **kwargs):
300			url = '/%s/%s' % (self.resource.get_url_path_name(), instance_id)
301			response = self.client.delete(url, **kwargs)
302			if response.status_code not in [200, 204, 404]:
303			self.handle_error(response)
304			return False
305			try:
306			resp_json = response.json()
307			if resp_json:
308			return resp_json
309			except:
310			pass
311			return True
312
313
314			class ActionAliasResourceManager(ResourceManager):
315			def __init__(self, resource, endpoint, cacert=None, debug=False):
			0 ignored issues – show Bug introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report The `__init__` method of the super-class `ResourceManager` is not called. It is generally advisable to initialize the super-class by calling its `__init__` method: class SomeParent: def __init__(self): self.x = 1 class SomeChild(SomeParent): def __init__(self): # Initialize the super class SomeParent.__init__(self) Loading history...
316			self.resource = resource
317			self.debug = debug
318			self.client = httpclient.HTTPClient(root=endpoint, cacert=cacert, debug=debug)
319
320
321			class LiveActionResourceManager(ResourceManager):
322			@add_auth_token_to_kwargs_from_env
323			def re_run(self, execution_id, parameters=None, **kwargs):
324			url = '/%s/%s/re_run' % (self.resource.get_url_path_name(), execution_id)
325
326			data = {}
327			if parameters:
328			data['parameters'] = parameters
329
330			response = self.client.post(url, data, **kwargs)
331			if response.status_code != 200:
332			self.handle_error(response)
333
334			instance = self.resource.deserialize(response.json())
335			return instance
336
337
338			class TriggerInstanceResourceManager(ResourceManager):
339			@add_auth_token_to_kwargs_from_env
340			def re_emit(self, trigger_instance_id, **kwargs):
341			url = '/%s/%s/re_emit' % (self.resource.get_url_path_name(), trigger_instance_id)
342			response = self.client.post(url, None)
343			if response.status_code != 200:
344			self.handle_error(response)
345			return response.json()
346

StackStorm / st2

Push — master ( bb5487...4fc1d0 )

st2client.models.WebhookBranch A

Complexity

Size/Duplication

Duplication Side-by-Side

Filter issues like