PackResourceManager.search() - Code Metrics - Inspection of "Pack management" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2842)

by Edward

created 2016-10-19 03:07 UTC

PackResourceManager.search() B

↳ Parent: PackResourceManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
cc	6
c	4
b	0
f	0
dl	0
loc	15
rs	8

# 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 sseclient import SSEClient

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)

    @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)
        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:
            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)

    @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,
            '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 PackResourceManager(ResourceManager):
    @add_auth_token_to_kwargs_from_env
    def install(self, packs, **kwargs):
        url = '/%s/install' % (self.resource.get_url_path_name())
        response = self.client.post(url, {'packs': packs}, **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 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 = self.resource.deserialize(response.json())
        return instance

    @add_auth_token_to_kwargs_from_env
    def create(self, args, **kwargs):
        url = '/%s/init' % (self.resource.get_url_path_name())
        keys = ['name', 'description', 'keywords', 'version', 'author',
                'email']
        payload = {}

        for key in keys:
            value = getattr(args, key)
            if value:
                payload[key] = value

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

StackStorm / st2

Pull Request — master (#2842)

PackResourceManager.search() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like