ResourceManager._query_details() - Code Metrics - Inspection of "Merge pull request #3612 from StackStorm/more_v2.3..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — plexxi-v2.3.2 ( 5d46fe )

unknown

created 2017-08-22 17:03 UTC

ResourceManager._query_details() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	11
c	1
b	0
f	0
dl	0
loc	33
rs	3.1764

How to fix 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 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')
        if not kwargs.get('api_key') and os.environ.get('ST2_API_KEY', None):
            kwargs['api_key'] = os.environ.get('ST2_API_KEY')

        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 = kwargs.pop('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 = kwargs.pop('token', None)
        api_key = kwargs.pop('api_key', None)

        if kwargs:
            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)

        if token:
            response = self.client.get(url, token=token)
        elif api_key:
            response = self.client.get(url, api_key=api_key)
        else:
            response = 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)

    def _query_details(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)
        api_key = kwargs.get('api_key', None)
        params = kwargs.get('params', {})

        for k, v in six.iteritems(kwargs):
            # Note: That's a special case to support api_key and token kwargs
            if k not in ['token', 'api_key']:
                params[k] = v

        url = '/%s/?%s' % (self.resource.get_url_path_name(),
                           urllib.parse.urlencode(params))

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

        if response.status_code == 404:
            # for query and query_with_count
            return [], None
        if response.status_code != 200:
            self.handle_error(response)
        items = response.json()
        instances = [self.resource.deserialize(item) for item in items]
        return instances, response

    @add_auth_token_to_kwargs_from_env
    def query(self, **kwargs):
        instances, _ = self._query_details(**kwargs)
        return instances

    @add_auth_token_to_kwargs_from_env
    def query_with_count(self, **kwargs):
        instances, response = self._query_details(**kwargs)
        if response and 'X-Total-Count' in response.headers:
            return (instances, int(response.headers['X-Total-Count']))
        else:
            return (instances, None)

    @add_auth_token_to_kwargs_from_env
    def get_by_name(self, name, **kwargs):
        instances = self.query(name=name, **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))
            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)

    @add_auth_token_to_kwargs_from_env
    def match(self, instance, **kwargs):
        url = '/%s/match' % self.resource.get_url_path_name()
        response = self.client.post(url, instance.serialize(), **kwargs)
        if response.status_code != 201:
            self.handle_error(response)
        matches = response.json()
        if len(matches) > 0:
            return (self.resource.deserialize(matches[0]['actionalias']),
                    matches[0]['representation'])
        else:
            return matches


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

        tasks = tasks or []
        no_reset = no_reset or []

        if list(set(no_reset) - set(tasks)):
            raise ValueError('List of tasks to reset does not match the tasks to rerun.')

        data = {
            'parameters': parameters or {},
            'tasks': tasks,
            'reset': list(set(tasks) - set(no_reset))
        }

        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, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        return response.json()


class AsyncRequest(Resource):
    pass


class PackResourceManager(ResourceManager):
    @add_auth_token_to_kwargs_from_env
    def install(self, packs, force=False, **kwargs):
        url = '/%s/install' % (self.resource.get_url_path_name())
        payload = {
            'packs': packs,
            'force': force
        }
        response = self.client.post(url, payload, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        instance = AsyncRequest.deserialize(response.json())
        return instance

    @add_auth_token_to_kwargs_from_env
    def remove(self, packs, **kwargs):
        url = '/%s/uninstall' % (self.resource.get_url_path_name())
        response = self.client.post(url, {'packs': packs}, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        instance = AsyncRequest.deserialize(response.json())
        return instance

    @add_auth_token_to_kwargs_from_env
    def search(self, args, **kwargs):
        url = '/%s/index/search' % (self.resource.get_url_path_name())
        if 'query' in vars(args):
            payload = {'query': args.query}
        else:
            payload = {'pack': args.pack}
        response = self.client.post(url, payload, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        data = response.json()
        if isinstance(data, list):
            return [self.resource.deserialize(item) for item in data]
        else:
            return self.resource.deserialize(data) if data else None

    @add_auth_token_to_kwargs_from_env
    def register(self, packs=None, types=None, **kwargs):
        url = '/%s/register' % (self.resource.get_url_path_name())
        payload = {}
        if types:
            payload['types'] = types
        if packs:
            payload['packs'] = packs
        response = self.client.post(url, payload, **kwargs)
        if response.status_code != 200:
            self.handle_error(response)
        instance = self.resource.deserialize(response.json())
        return instance


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


class StreamManager(object):
    def __init__(self, endpoint, cacert, debug):
        self._url = httpclient.get_url_without_trailing_slash(endpoint) + '/stream'
        self.debug = debug
        self.cacert = cacert

    @add_auth_token_to_kwargs_from_env
    def listen(self, events, **kwargs):
        # Late import to avoid very expensive in-direct import (~1 second) when this function is
        # not called / used
        from sseclient import SSEClient

        url = self._url

        if 'token' in kwargs:
            url += '?x-auth-token=%s' % kwargs.get('token')

        if 'api_key' in kwargs:
            url += '?st2-api-key=%s' % kwargs.get('api_key')

        if isinstance(events, six.string_types):
            events = [events]

        for message in SSEClient(url):
            if message.event in events:
                yield json.loads(message.data)


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
25			from st2client.utils import httpclient
26
27
28			LOG = logging.getLogger(__name__)
29
30
31			def add_auth_token_to_kwargs_from_env(func):
32			@wraps(func)
33			def decorate(args, *kwargs):
34			if not kwargs.get('token') and os.environ.get('ST2_AUTH_TOKEN', None):
35			kwargs['token'] = os.environ.get('ST2_AUTH_TOKEN')
36			if not kwargs.get('api_key') and os.environ.get('ST2_API_KEY', None):
37			kwargs['api_key'] = os.environ.get('ST2_API_KEY')
38
39			return func(args, *kwargs)
40			return decorate
41
42
43			class Resource(object):
44
45			# An alias to use for the resource if different than the class name.
46			_alias = None
47
48			# Display name of the resource. This may be different than its resource
49			# name specifically when the resource name is composed of multiple words.
50			_display_name = None
51
52			# URL path for the resource.
53			_url_path = None
54
55			# Plural form of the resource name. This will be used to build the
56			# latter part of the REST URL.
57			_plural = None
58
59			# Plural form of the resource display name.
60			_plural_display_name = None
61
62			# A list of class attributes which will be included in __repr__ return value
63			_repr_attributes = []
64
65			def __init__(self, args, *kwargs):
66			for k, v in six.iteritems(kwargs):
67			setattr(self, k, v)
68
69			def to_dict(self, exclude_attributes=None):
70			"""
71			Return a dictionary representation of this object.
72
73			:param exclude_attributes: Optional list of attributes to exclude.
74			:type exclude_attributes: ``list``
75
76			:rtype: ``dict``
77			"""
78			exclude_attributes = exclude_attributes or []
79
80			attributes = self.__dict__.keys()
81			attributes = [attr for attr in attributes if not attr.startswith('__') and
82			attr not in exclude_attributes]
83
84			result = {}
85			for attribute in attributes:
86			value = getattr(self, attribute, None)
87			result[attribute] = value
88
89			return result
90
91			@classmethod
92			def get_alias(cls):
93			return cls._alias if cls._alias else cls.__name__
94
95			@classmethod
96			def get_display_name(cls):
97			return cls._display_name if cls._display_name else cls.__name__
98
99			@classmethod
100			def get_plural_name(cls):
101			if not cls._plural:
102			raise Exception('The %s class is missing class attributes '
103			'in its definition.' % cls.__name__)
104			return cls._plural
105
106			@classmethod
107			def get_plural_display_name(cls):
108			return (cls._plural_display_name
109			if cls._plural_display_name
110			else cls._plural)
111
112			@classmethod
113			def get_url_path_name(cls):
114			if cls._url_path:
115			return cls._url_path
116
117			return cls.get_plural_name().lower()
118
119			def serialize(self):
120			return dict((k, v)
121			for k, v in six.iteritems(self.__dict__)
122			if not k.startswith('_'))
123
124			@classmethod
125			def deserialize(cls, doc):
126			if type(doc) is not dict:
127			doc = json.loads(doc)
128			return cls(**doc)
129
130			def __str__(self):
131			return str(self.__repr__())
132
133			def __repr__(self):
134			if not self._repr_attributes:
135			return super(Resource, self).__repr__()
136
137			attributes = []
138			for attribute in self._repr_attributes:
139			value = getattr(self, attribute, None)
140			attributes.append('%s=%s' % (attribute, value))
141
142			attributes = ','.join(attributes)
143			class_name = self.__class__.__name__
144			result = '<%s %s>' % (class_name, attributes)
145			return result
146
147
148			class ResourceManager(object):
149
150			def __init__(self, resource, endpoint, cacert=None, debug=False):
151			self.resource = resource
152			self.debug = debug
153			self.client = httpclient.HTTPClient(endpoint, cacert=cacert, debug=debug)
154
155			@staticmethod
156			def handle_error(response):
157			try:
158			content = response.json()
159			fault = content.get('faultstring', '') if content else ''
160			if fault:
161			response.reason += '\nMESSAGE: %s' % fault
162			except Exception as e:
163			response.reason += ('\nUnable to retrieve detailed message '
164			'from the HTTP response. %s\n' % str(e))
165			response.raise_for_status()
166
167			@add_auth_token_to_kwargs_from_env
168			def get_all(self, **kwargs):
169			# TODO: This is ugly, stop abusing kwargs
170			url = '/%s' % self.resource.get_url_path_name()
171			limit = kwargs.pop('limit', None)
172			pack = kwargs.pop('pack', None)
173			prefix = kwargs.pop('prefix', None)
174			user = kwargs.pop('user', None)
175
176			params = kwargs.pop('params', {})
177			if limit and limit <= 0:
178			limit = None
179			if limit:
180			params['limit'] = limit
181
182			if pack:
183			params['pack'] = pack
184
185			if prefix:
186			params['prefix'] = prefix
187
188			if user:
189			params['user'] = user
190
191			response = self.client.get(url=url, params=params, **kwargs)
192			if response.status_code != 200:
193			self.handle_error(response)
194			return [self.resource.deserialize(item)
195			for item in response.json()]
196
197			@add_auth_token_to_kwargs_from_env
198			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...
199			url = '/%s/%s' % (self.resource.get_url_path_name(), id)
200			response = self.client.get(url, **kwargs)
201			if response.status_code == 404:
202			return None
203			if response.status_code != 200:
204			self.handle_error(response)
205			return self.resource.deserialize(response.json())
206
207			@add_auth_token_to_kwargs_from_env
208			def get_property(self, id_, property_name, self_deserialize=True, **kwargs):
209			"""
210			Gets a property of a Resource.
211			id_ : Id of the resource
212			property_name: Name of the property
213			self_deserialize: #Implies use the deserialize method implemented by this resource.
214			"""
215			token = kwargs.pop('token', None)
216			api_key = kwargs.pop('api_key', None)
217
218			if kwargs:
219			url = '/%s/%s/%s/?%s' % (self.resource.get_url_path_name(), id_, property_name,
220			urllib.parse.urlencode(kwargs))
221			else:
222			url = '/%s/%s/%s/' % (self.resource.get_url_path_name(), id_, property_name)
223
224			if token:
225			response = self.client.get(url, token=token)
226			elif api_key:
227			response = self.client.get(url, api_key=api_key)
228			else:
229			response = self.client.get(url)
230
231			if response.status_code == 404:
232			return None
233			if response.status_code != 200:
234			self.handle_error(response)
235
236			if self_deserialize:
237			return [self.resource.deserialize(item) for item in response.json()]
238			else:
239			return response.json()
240
241			@add_auth_token_to_kwargs_from_env
242			def get_by_ref_or_id(self, ref_or_id, **kwargs):
243			return self.get_by_id(id=ref_or_id, **kwargs)
244
245			def _query_details(self, **kwargs):
246			if not kwargs:
247			raise Exception('Query parameter is not provided.')
248			if 'limit' in kwargs and kwargs.get('limit') <= 0:
249			kwargs.pop('limit')
250
251			token = kwargs.get('token', None)
252			api_key = kwargs.get('api_key', None)
253			params = kwargs.get('params', {})
254
255			for k, v in six.iteritems(kwargs):
256			# Note: That's a special case to support api_key and token kwargs
257			if k not in ['token', 'api_key']:
258			params[k] = v
259
260			url = '/%s/?%s' % (self.resource.get_url_path_name(),
261			urllib.parse.urlencode(params))
262
263			if token:
264			response = self.client.get(url, token=token)
265			elif api_key:
266			response = self.client.get(url, api_key=api_key)
267			else:
268			response = self.client.get(url)
269
270			if response.status_code == 404:
271			# for query and query_with_count
272			return [], None
273			if response.status_code != 200:
274			self.handle_error(response)
275			items = response.json()
276			instances = [self.resource.deserialize(item) for item in items]
277			return instances, response
278
279			@add_auth_token_to_kwargs_from_env
280			def query(self, **kwargs):
281			instances, _ = self._query_details(**kwargs)
282			return instances
283
284			@add_auth_token_to_kwargs_from_env
285			def query_with_count(self, **kwargs):
286			instances, response = self._query_details(**kwargs)
287			if response and 'X-Total-Count' in response.headers:
288			return (instances, int(response.headers['X-Total-Count']))
289			else:
290			return (instances, None)
291
292			@add_auth_token_to_kwargs_from_env
293			def get_by_name(self, name, **kwargs):
294			instances = self.query(name=name, **kwargs)
295			if not instances:
296			return None
297			else:
298			if len(instances) > 1:
299			raise Exception('More than one %s named "%s" are found.' %
300			(self.resource.__name__.lower(), name))
301			return instances[0]
302
303			@add_auth_token_to_kwargs_from_env
304			def create(self, instance, **kwargs):
305			url = '/%s' % self.resource.get_url_path_name()
306			response = self.client.post(url, instance.serialize(), **kwargs)
307			if response.status_code != 200:
308			self.handle_error(response)
309			instance = self.resource.deserialize(response.json())
310			return instance
311
312			@add_auth_token_to_kwargs_from_env
313			def update(self, instance, **kwargs):
314			url = '/%s/%s' % (self.resource.get_url_path_name(), instance.id)
315			response = self.client.put(url, instance.serialize(), **kwargs)
316			if response.status_code != 200:
317			self.handle_error(response)
318			instance = self.resource.deserialize(response.json())
319			return instance
320
321			@add_auth_token_to_kwargs_from_env
322			def delete(self, instance, **kwargs):
323			url = '/%s/%s' % (self.resource.get_url_path_name(), instance.id)
324			response = self.client.delete(url, **kwargs)
325
326			if response.status_code not in [200, 204, 404]:
327			self.handle_error(response)
328			return False
329
330			return True
331
332			@add_auth_token_to_kwargs_from_env
333			def delete_by_id(self, instance_id, **kwargs):
334			url = '/%s/%s' % (self.resource.get_url_path_name(), instance_id)
335			response = self.client.delete(url, **kwargs)
336			if response.status_code not in [200, 204, 404]:
337			self.handle_error(response)
338			return False
339			try:
340			resp_json = response.json()
341			if resp_json:
342			return resp_json
343			except:
344			pass
345			return True
346
347
348			class ActionAliasResourceManager(ResourceManager):
349			def __init__(self, resource, endpoint, cacert=None, debug=False):
			0 ignored issues – show Bug introduced 2016-01-14 18:36 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...
350			self.resource = resource
351			self.debug = debug
352			self.client = httpclient.HTTPClient(root=endpoint, cacert=cacert, debug=debug)
353
354			@add_auth_token_to_kwargs_from_env
355			def match(self, instance, **kwargs):
356			url = '/%s/match' % self.resource.get_url_path_name()
357			response = self.client.post(url, instance.serialize(), **kwargs)
358			if response.status_code != 201:
359			self.handle_error(response)
360			matches = response.json()
361			if len(matches) > 0:
362			return (self.resource.deserialize(matches[0]['actionalias']),
363			matches[0]['representation'])
364			else:
365			return matches
366
367
368			class LiveActionResourceManager(ResourceManager):
369			@add_auth_token_to_kwargs_from_env
370			def re_run(self, execution_id, parameters=None, tasks=None, no_reset=None, **kwargs):
371			url = '/%s/%s/re_run' % (self.resource.get_url_path_name(), execution_id)
372
373			tasks = tasks or []
374			no_reset = no_reset or []
375
376			if list(set(no_reset) - set(tasks)):
377			raise ValueError('List of tasks to reset does not match the tasks to rerun.')
378
379			data = {
380			'parameters': parameters or {},
381			'tasks': tasks,
382			'reset': list(set(tasks) - set(no_reset))
383			}
384
385			response = self.client.post(url, data, **kwargs)
386			if response.status_code != 200:
387			self.handle_error(response)
388
389			instance = self.resource.deserialize(response.json())
390			return instance
391
392
393			class TriggerInstanceResourceManager(ResourceManager):
394			@add_auth_token_to_kwargs_from_env
395			def re_emit(self, trigger_instance_id, **kwargs):
396			url = '/%s/%s/re_emit' % (self.resource.get_url_path_name(), trigger_instance_id)
397			response = self.client.post(url, None, **kwargs)
398			if response.status_code != 200:
399			self.handle_error(response)
400			return response.json()
401
402
403			class AsyncRequest(Resource):
404			pass
405
406
407			class PackResourceManager(ResourceManager):
408			@add_auth_token_to_kwargs_from_env
409			def install(self, packs, force=False, **kwargs):
410			url = '/%s/install' % (self.resource.get_url_path_name())
411			payload = {
412			'packs': packs,
413			'force': force
414			}
415			response = self.client.post(url, payload, **kwargs)
416			if response.status_code != 200:
417			self.handle_error(response)
418			instance = AsyncRequest.deserialize(response.json())
419			return instance
420
421			@add_auth_token_to_kwargs_from_env
422			def remove(self, packs, **kwargs):
423			url = '/%s/uninstall' % (self.resource.get_url_path_name())
424			response = self.client.post(url, {'packs': packs}, **kwargs)
425			if response.status_code != 200:
426			self.handle_error(response)
427			instance = AsyncRequest.deserialize(response.json())
428			return instance
429
430			@add_auth_token_to_kwargs_from_env
431			def search(self, args, **kwargs):
432			url = '/%s/index/search' % (self.resource.get_url_path_name())
433			if 'query' in vars(args):
434			payload = {'query': args.query}
435			else:
436			payload = {'pack': args.pack}
437			response = self.client.post(url, payload, **kwargs)
438			if response.status_code != 200:
439			self.handle_error(response)
440			data = response.json()
441			if isinstance(data, list):
442			return [self.resource.deserialize(item) for item in data]
443			else:
444			return self.resource.deserialize(data) if data else None
445
446			@add_auth_token_to_kwargs_from_env
447			def register(self, packs=None, types=None, **kwargs):
448			url = '/%s/register' % (self.resource.get_url_path_name())
449			payload = {}
450			if types:
451			payload['types'] = types
452			if packs:
453			payload['packs'] = packs
454			response = self.client.post(url, payload, **kwargs)
455			if response.status_code != 200:
456			self.handle_error(response)
457			instance = self.resource.deserialize(response.json())
458			return instance
459
460
461			class ConfigManager(ResourceManager):
462			@add_auth_token_to_kwargs_from_env
463			def update(self, instance, **kwargs):
464			url = '/%s/%s' % (self.resource.get_url_path_name(), instance.pack)
465			response = self.client.put(url, instance.values, **kwargs)
466			if response.status_code != 200:
467			self.handle_error(response)
468			instance = self.resource.deserialize(response.json())
469			return instance
470
471
472			class StreamManager(object):
473			def __init__(self, endpoint, cacert, debug):
474			self._url = httpclient.get_url_without_trailing_slash(endpoint) + '/stream'
475			self.debug = debug
476			self.cacert = cacert
477
478			@add_auth_token_to_kwargs_from_env
479			def listen(self, events, **kwargs):
480			# Late import to avoid very expensive in-direct import (~1 second) when this function is
481			# not called / used
482			from sseclient import SSEClient
483
484			url = self._url
485
486			if 'token' in kwargs:
487			url += '?x-auth-token=%s' % kwargs.get('token')
488
489			if 'api_key' in kwargs:
490			url += '?st2-api-key=%s' % kwargs.get('api_key')
491
492			if isinstance(events, six.string_types):
493			events = [events]
494
495			for message in SSEClient(url):
496			if message.event in events:
497			yield json.loads(message.data)
498

Plexxi / st2

GitHub Access Token became invalid

Push — plexxi-v2.3.2 ( 5d46fe )

ResourceManager._query_details() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like