Access._get_publisher() - Code Metrics - Inspection of "log_files_paths -> log_file_paths" - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

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

unknown

created 2016-04-28 20:26 UTC

Access._get_publisher() A

↳ Parent: Access

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	4
rs	10

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import abc

import six
from mongoengine import NotUniqueError

from st2common import log as logging
from st2common.exceptions.db import StackStormDBObjectConflictError
from st2common.models.system.common import ResourceReference
from st2common.transport.reactor import TriggerDispatcher


__all__ = [
    'Access',

    'ContentPackResource',
    'StatusBasedResource'
]

LOG = logging.getLogger(__name__)


@six.add_metaclass(abc.ABCMeta)
class Access(object):
    impl = None
    publisher = None
    dispatcher = None

    # ModelAPI class for this resource
    api_model_cls = None

    # A list of operations for which we should dispatch a trigger
    dispatch_trigger_for_operations = []

    # Maps model operation name (e.g. create, update, delete) to the trigger reference which is
    # used when dispatching a trigger
    operation_to_trigger_ref_map = {}

    @classmethod
    @abc.abstractmethod
    def _get_impl(cls):
        pass

    @classmethod
    @abc.abstractmethod
    def _get_publisher(cls):
        return None

    @classmethod
    def _get_dispatcher(cls):
        """
        Return a dispatcher class which is used for dispatching triggers.
        """
        if not cls.dispatcher:
            cls.dispatcher = TriggerDispatcher(LOG)

        return cls.dispatcher

    @classmethod
    @abc.abstractmethod
    def _get_by_object(cls, object):

        return None

    @classmethod
    def get_by_name(cls, value):
        return cls._get_impl().get_by_name(value)

    @classmethod
    def get_by_id(cls, value):
        return cls._get_impl().get_by_id(value)

    @classmethod
    def get_by_ref(cls, value):
        return cls._get_impl().get_by_ref(value)

    @classmethod
    def get(cls, *args, **kwargs):
        return cls._get_impl().get(*args, **kwargs)

    @classmethod
    def get_all(cls, *args, **kwargs):
        return cls._get_impl().get_all(*args, **kwargs)

    @classmethod
    def count(cls, *args, **kwargs):
        return cls._get_impl().count(*args, **kwargs)

    @classmethod
    def query(cls, *args, **kwargs):
        return cls._get_impl().query(*args, **kwargs)

    @classmethod
    def distinct(cls, *args, **kwargs):
        return cls._get_impl().distinct(*args, **kwargs)

    @classmethod
    def aggregate(cls, *args, **kwargs):
        return cls._get_impl().aggregate(*args, **kwargs)

    @classmethod
    def insert(cls, model_object, publish=True, dispatch_trigger=True,
               log_not_unique_error_as_debug=False):
        if model_object.id:
            raise ValueError('id for object %s was unexpected.' % model_object)
        try:
            model_object = cls._get_impl().insert(model_object)
        except NotUniqueError as e:
            if log_not_unique_error_as_debug:
                LOG.debug('Conflict while trying to save in DB.', exc_info=True)
            else:
                LOG.exception('Conflict while trying to save in DB.')
            # On a conflict determine the conflicting object and return its id in
            # the raised exception.
            conflict_object = cls._get_by_object(model_object)
            conflict_id = str(conflict_object.id) if conflict_object else None
            message = str(e)
            raise StackStormDBObjectConflictError(message=message, conflict_id=conflict_id,
                                                  model_object=model_object)

        # Publish internal event on the message bus
        if publish:
            try:
                cls.publish_create(model_object)
            except:
                LOG.exception('Publish failed.')

        # Dispatch trigger
        if dispatch_trigger:
            try:
                cls.dispatch_create_trigger(model_object)
            except:
                LOG.exception('Trigger dispatch failed.')

        return model_object

    @classmethod
    def add_or_update(cls, model_object, publish=True, dispatch_trigger=True,
                      log_not_unique_error_as_debug=False):
        pre_persist_id = model_object.id
        try:
            model_object = cls._get_impl().add_or_update(model_object)
        except NotUniqueError as e:
            if log_not_unique_error_as_debug:
                LOG.debug('Conflict while trying to save in DB.', exc_info=True)
            else:
                LOG.exception('Conflict while trying to save in DB.')
            # On a conflict determine the conflicting object and return its id in
            # the raised exception.
            conflict_object = cls._get_by_object(model_object)
            conflict_id = str(conflict_object.id) if conflict_object else None
            message = str(e)
            raise StackStormDBObjectConflictError(message=message, conflict_id=conflict_id,
                                                  model_object=model_object)

        is_update = str(pre_persist_id) == str(model_object.id)

        # Publish internal event on the message bus
        if publish:
            try:
                if is_update:
                    cls.publish_update(model_object)
                else:
                    cls.publish_create(model_object)
            except:
                LOG.exception('Publish failed.')

        # Dispatch trigger
        if dispatch_trigger:
            try:
                if is_update:
                    cls.dispatch_update_trigger(model_object)
                else:
                    cls.dispatch_create_trigger(model_object)
            except:
                LOG.exception('Trigger dispatch failed.')

        return model_object

    @classmethod
    def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs):
        """
        Use this method when -
        * upsert=False is desired
        * special operators like push, push_all are to be used.
        """
        cls._get_impl().update(model_object, **kwargs)
        # update does not return the object but a flag; likely success/fail but docs
        # are not very good on this one so ignoring. Explicitly get the object from
        # DB abd return.
        model_object = cls.get_by_id(model_object.id)

        # Publish internal event on the message bus
        if publish:
            try:
                cls.publish_update(model_object)
            except:
                LOG.exception('Publish failed.')

        # Dispatch trigger
        if dispatch_trigger:
            try:
                cls.dispatch_update_trigger(model_object)
            except:
                LOG.exception('Trigger dispatch failed.')

        return model_object

    @classmethod
    def delete(cls, model_object, publish=True, dispatch_trigger=True):
        persisted_object = cls._get_impl().delete(model_object)

        # Publish internal event on the message bus
        if publish:
            try:
                cls.publish_delete(model_object)
            except Exception:
                LOG.exception('Publish failed.')

        # Dispatch trigger
        if dispatch_trigger:
            try:
                cls.dispatch_delete_trigger(model_object)
            except Exception:
                LOG.exception('Trigger dispatch failed.')

        return persisted_object

    ####################################################
    # Internal event bus message publish related methods
    ####################################################

    @classmethod
    def publish_create(cls, model_object):
        publisher = cls._get_publisher()
        if publisher:
            publisher.publish_create(model_object)

    @classmethod
    def publish_update(cls, model_object):
        publisher = cls._get_publisher()
        if publisher:
            publisher.publish_update(model_object)

    @classmethod
    def publish_delete(cls, model_object):
        publisher = cls._get_publisher()
        if publisher:
            publisher.publish_delete(model_object)

    ############################################
    # Internal trigger dispatch related methods
    ###########################################

    @classmethod
    def dispatch_create_trigger(cls, model_object):
        """
        Dispatch a resource-specific trigger which indicates a new resource has been created.
        """
        return cls._dispatch_operation_trigger(operation='create', model_object=model_object)

    @classmethod
    def dispatch_update_trigger(cls, model_object):
        """
        Dispatch a resource-specific trigger which indicates an existing resource has been updated.
        """
        return cls._dispatch_operation_trigger(operation='update', model_object=model_object)

    @classmethod
    def dispatch_delete_trigger(cls, model_object):
        """
        Dispatch a resource-specific trigger which indicates an existing resource has been
        deleted.
        """
        return cls._dispatch_operation_trigger(operation='delete', model_object=model_object)

    @classmethod
    def _get_trigger_ref_for_operation(cls, operation):
        trigger_ref = cls.operation_to_trigger_ref_map.get(operation, None)

        if not trigger_ref:
            raise ValueError('Trigger ref not specified for operation: %s' % (operation))

        return trigger_ref

    @classmethod
    def _dispatch_operation_trigger(cls, operation, model_object):
        if operation not in cls.dispatch_trigger_for_operations:
            return

        trigger = cls._get_trigger_ref_for_operation(operation=operation)

        object_payload = cls.api_model_cls.from_model(model_object, mask_secrets=True).__json__()
        payload = {
            'object': object_payload
        }
        return cls._dispatch_trigger(operation=operation, trigger=trigger, payload=payload)

    @classmethod
    def _dispatch_trigger(cls, operation, trigger, payload):
        if operation not in cls.dispatch_trigger_for_operations:
            return

        dispatcher = cls._get_dispatcher()
        return dispatcher.dispatch(trigger=trigger, payload=payload)


class ContentPackResource(Access):

    @classmethod
    def get_by_ref(cls, ref):
        if not ref:
            return None

        ref_obj = ResourceReference.from_string_reference(ref=ref)
        result = cls.query(name=ref_obj.name,
                           pack=ref_obj.pack).first()
        return result

    @classmethod
    def _get_by_object(cls, object):

        # For an object with a resourcepack pack.name is unique.
        name = getattr(object, 'name', '')
        pack = getattr(object, 'pack', '')
        return cls.get_by_ref(ResourceReference.to_string_reference(pack=pack, name=name))


class StatusBasedResource(Access):
    """Persistence layer for models that needs to publish status to the message queue."""

    @classmethod
    def publish_status(cls, model_object):
        """Publish the object status to the message queue.

        Publish the instance of the model as payload with the status
        as routing key to the message queue via the StatePublisher.

        :param model_object: An instance of the model.
        :type model_object: ``object``
        """
        publisher = cls._get_publisher()
        if publisher:
            publisher.publish_state(model_object, getattr(model_object, 'status', None))


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import abc
17
18			import six
19			from mongoengine import NotUniqueError
20
21			from st2common import log as logging
22			from st2common.exceptions.db import StackStormDBObjectConflictError
23			from st2common.models.system.common import ResourceReference
24			from st2common.transport.reactor import TriggerDispatcher
25
26
27			__all__ = [
28			'Access',
29
30			'ContentPackResource',
31			'StatusBasedResource'
32			]
33
34			LOG = logging.getLogger(__name__)
35
36
37			@six.add_metaclass(abc.ABCMeta)
38			class Access(object):
39			impl = None
40			publisher = None
41			dispatcher = None
42
43			# ModelAPI class for this resource
44			api_model_cls = None
45
46			# A list of operations for which we should dispatch a trigger
47			dispatch_trigger_for_operations = []
48
49			# Maps model operation name (e.g. create, update, delete) to the trigger reference which is
50			# used when dispatching a trigger
51			operation_to_trigger_ref_map = {}
52
53			@classmethod
54			@abc.abstractmethod
55			def _get_impl(cls):
56			pass
57
58			@classmethod
59			@abc.abstractmethod
60			def _get_publisher(cls):
61			return None
62
63			@classmethod
64			def _get_dispatcher(cls):
65			"""
66			Return a dispatcher class which is used for dispatching triggers.
67			"""
68			if not cls.dispatcher:
69			cls.dispatcher = TriggerDispatcher(LOG)
70
71			return cls.dispatcher
72
73			@classmethod
74			@abc.abstractmethod
75			def _get_by_object(cls, object):
			0 ignored issues – show Bug Best Practice introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `object`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
76			return None
77
78			@classmethod
79			def get_by_name(cls, value):
80			return cls._get_impl().get_by_name(value)
81
82			@classmethod
83			def get_by_id(cls, value):
84			return cls._get_impl().get_by_id(value)
85
86			@classmethod
87			def get_by_ref(cls, value):
88			return cls._get_impl().get_by_ref(value)
89
90			@classmethod
91			def get(cls, args, *kwargs):
92			return cls._get_impl().get(args, *kwargs)
93
94			@classmethod
95			def get_all(cls, args, *kwargs):
96			return cls._get_impl().get_all(args, *kwargs)
97
98			@classmethod
99			def count(cls, args, *kwargs):
100			return cls._get_impl().count(args, *kwargs)
101
102			@classmethod
103			def query(cls, args, *kwargs):
104			return cls._get_impl().query(args, *kwargs)
105
106			@classmethod
107			def distinct(cls, args, *kwargs):
108			return cls._get_impl().distinct(args, *kwargs)
109
110			@classmethod
111			def aggregate(cls, args, *kwargs):
112			return cls._get_impl().aggregate(args, *kwargs)
113
114			@classmethod
115			def insert(cls, model_object, publish=True, dispatch_trigger=True,
116			log_not_unique_error_as_debug=False):
117			if model_object.id:
118			raise ValueError('id for object %s was unexpected.' % model_object)
119			try:
120			model_object = cls._get_impl().insert(model_object)
121			except NotUniqueError as e:
122			if log_not_unique_error_as_debug:
123			LOG.debug('Conflict while trying to save in DB.', exc_info=True)
124			else:
125			LOG.exception('Conflict while trying to save in DB.')
126			# On a conflict determine the conflicting object and return its id in
127			# the raised exception.
128			conflict_object = cls._get_by_object(model_object)
129			conflict_id = str(conflict_object.id) if conflict_object else None
130			message = str(e)
131			raise StackStormDBObjectConflictError(message=message, conflict_id=conflict_id,
132			model_object=model_object)
133
134			# Publish internal event on the message bus
135			if publish:
136			try:
137			cls.publish_create(model_object)
138			except:
139			LOG.exception('Publish failed.')
140
141			# Dispatch trigger
142			if dispatch_trigger:
143			try:
144			cls.dispatch_create_trigger(model_object)
145			except:
146			LOG.exception('Trigger dispatch failed.')
147
148			return model_object
149
150			@classmethod
151			def add_or_update(cls, model_object, publish=True, dispatch_trigger=True,
152			log_not_unique_error_as_debug=False):
153			pre_persist_id = model_object.id
154			try:
155			model_object = cls._get_impl().add_or_update(model_object)
156			except NotUniqueError as e:
157			if log_not_unique_error_as_debug:
158			LOG.debug('Conflict while trying to save in DB.', exc_info=True)
159			else:
160			LOG.exception('Conflict while trying to save in DB.')
161			# On a conflict determine the conflicting object and return its id in
162			# the raised exception.
163			conflict_object = cls._get_by_object(model_object)
164			conflict_id = str(conflict_object.id) if conflict_object else None
165			message = str(e)
166			raise StackStormDBObjectConflictError(message=message, conflict_id=conflict_id,
167			model_object=model_object)
168
169			is_update = str(pre_persist_id) == str(model_object.id)
170
171			# Publish internal event on the message bus
172			if publish:
173			try:
174			if is_update:
175			cls.publish_update(model_object)
176			else:
177			cls.publish_create(model_object)
178			except:
179			LOG.exception('Publish failed.')
180
181			# Dispatch trigger
182			if dispatch_trigger:
183			try:
184			if is_update:
185			cls.dispatch_update_trigger(model_object)
186			else:
187			cls.dispatch_create_trigger(model_object)
188			except:
189			LOG.exception('Trigger dispatch failed.')
190
191			return model_object
192
193			@classmethod
194			def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs):
195			"""
196			Use this method when -
197			* upsert=False is desired
198			* special operators like push, push_all are to be used.
199			"""
200			cls._get_impl().update(model_object, **kwargs)
201			# update does not return the object but a flag; likely success/fail but docs
202			# are not very good on this one so ignoring. Explicitly get the object from
203			# DB abd return.
204			model_object = cls.get_by_id(model_object.id)
205
206			# Publish internal event on the message bus
207			if publish:
208			try:
209			cls.publish_update(model_object)
210			except:
211			LOG.exception('Publish failed.')
212
213			# Dispatch trigger
214			if dispatch_trigger:
215			try:
216			cls.dispatch_update_trigger(model_object)
217			except:
218			LOG.exception('Trigger dispatch failed.')
219
220			return model_object
221
222			@classmethod
223			def delete(cls, model_object, publish=True, dispatch_trigger=True):
224			persisted_object = cls._get_impl().delete(model_object)
225
226			# Publish internal event on the message bus
227			if publish:
228			try:
229			cls.publish_delete(model_object)
230			except Exception:
231			LOG.exception('Publish failed.')
232
233			# Dispatch trigger
234			if dispatch_trigger:
235			try:
236			cls.dispatch_delete_trigger(model_object)
237			except Exception:
238			LOG.exception('Trigger dispatch failed.')
239
240			return persisted_object
241
242			####################################################
243			# Internal event bus message publish related methods
244			####################################################
245
246			@classmethod
247			def publish_create(cls, model_object):
248			publisher = cls._get_publisher()
249			if publisher:
250			publisher.publish_create(model_object)
251
252			@classmethod
253			def publish_update(cls, model_object):
254			publisher = cls._get_publisher()
255			if publisher:
256			publisher.publish_update(model_object)
257
258			@classmethod
259			def publish_delete(cls, model_object):
260			publisher = cls._get_publisher()
261			if publisher:
262			publisher.publish_delete(model_object)
263
264			############################################
265			# Internal trigger dispatch related methods
266			###########################################
267
268			@classmethod
269			def dispatch_create_trigger(cls, model_object):
270			"""
271			Dispatch a resource-specific trigger which indicates a new resource has been created.
272			"""
273			return cls._dispatch_operation_trigger(operation='create', model_object=model_object)
274
275			@classmethod
276			def dispatch_update_trigger(cls, model_object):
277			"""
278			Dispatch a resource-specific trigger which indicates an existing resource has been updated.
279			"""
280			return cls._dispatch_operation_trigger(operation='update', model_object=model_object)
281
282			@classmethod
283			def dispatch_delete_trigger(cls, model_object):
284			"""
285			Dispatch a resource-specific trigger which indicates an existing resource has been
286			deleted.
287			"""
288			return cls._dispatch_operation_trigger(operation='delete', model_object=model_object)
289
290			@classmethod
291			def _get_trigger_ref_for_operation(cls, operation):
292			trigger_ref = cls.operation_to_trigger_ref_map.get(operation, None)
293
294			if not trigger_ref:
295			raise ValueError('Trigger ref not specified for operation: %s' % (operation))
296
297			return trigger_ref
298
299			@classmethod
300			def _dispatch_operation_trigger(cls, operation, model_object):
301			if operation not in cls.dispatch_trigger_for_operations:
302			return
303
304			trigger = cls._get_trigger_ref_for_operation(operation=operation)
305
306			object_payload = cls.api_model_cls.from_model(model_object, mask_secrets=True).__json__()
307			payload = {
308			'object': object_payload
309			}
310			return cls._dispatch_trigger(operation=operation, trigger=trigger, payload=payload)
311
312			@classmethod
313			def _dispatch_trigger(cls, operation, trigger, payload):
314			if operation not in cls.dispatch_trigger_for_operations:
315			return
316
317			dispatcher = cls._get_dispatcher()
318			return dispatcher.dispatch(trigger=trigger, payload=payload)
319
320
321			class ContentPackResource(Access):
322
323			@classmethod
324			def get_by_ref(cls, ref):
325			if not ref:
326			return None
327
328			ref_obj = ResourceReference.from_string_reference(ref=ref)
329			result = cls.query(name=ref_obj.name,
330			pack=ref_obj.pack).first()
331			return result
332
333			@classmethod
334			def _get_by_object(cls, object):
			0 ignored issues – show Bug Best Practice introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `object`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
335			# For an object with a resourcepack pack.name is unique.
336			name = getattr(object, 'name', '')
337			pack = getattr(object, 'pack', '')
338			return cls.get_by_ref(ResourceReference.to_string_reference(pack=pack, name=name))
339
340
341			class StatusBasedResource(Access):
342			"""Persistence layer for models that needs to publish status to the message queue."""
343
344			@classmethod
345			def publish_status(cls, model_object):
346			"""Publish the object status to the message queue.
347
348			Publish the instance of the model as payload with the status
349			as routing key to the message queue via the StatePublisher.
350
351			:param model_object: An instance of the model.
352			:type model_object: ``object``
353			"""
354			publisher = cls._get_publisher()
355			if publisher:
356			publisher.publish_state(model_object, getattr(model_object, 'status', None))
357

Plexxi / st2

GitHub Access Token became invalid

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

Access._get_publisher() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like