Access.add_or_update() - Code Metrics - Inspection of "Merge PR 3081 for PXPSE-1163" - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop-v1.6.0 ( 9d5181...7efb31 )

unknown

created 2016-12-02 22:05 UTC

Access.add_or_update() F

↳ Parent: Access

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
dl	0
loc	42
rs	3.1304
c	0
b	0
f	0

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 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_uid(cls, value):
        return cls._get_impl().get_by_uid(value)

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

    @classmethod
    def get_by_pack(cls, value):
        return cls._get_impl().get_by_pack(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: %s.', str(e))
            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: %s.', str(e))
            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_uid(cls, value):
88			return cls._get_impl().get_by_uid(value)
89
90			@classmethod
91			def get_by_ref(cls, value):
92			return cls._get_impl().get_by_ref(value)
93
94			@classmethod
95			def get_by_pack(cls, value):
96			return cls._get_impl().get_by_pack(value)
97
98			@classmethod
99			def get(cls, args, *kwargs):
100			return cls._get_impl().get(args, *kwargs)
101
102			@classmethod
103			def get_all(cls, args, *kwargs):
104			return cls._get_impl().get_all(args, *kwargs)
105
106			@classmethod
107			def count(cls, args, *kwargs):
108			return cls._get_impl().count(args, *kwargs)
109
110			@classmethod
111			def query(cls, args, *kwargs):
112			return cls._get_impl().query(args, *kwargs)
113
114			@classmethod
115			def distinct(cls, args, *kwargs):
116			return cls._get_impl().distinct(args, *kwargs)
117
118			@classmethod
119			def aggregate(cls, args, *kwargs):
120			return cls._get_impl().aggregate(args, *kwargs)
121
122			@classmethod
123			def insert(cls, model_object, publish=True, dispatch_trigger=True,
124			log_not_unique_error_as_debug=False):
125			if model_object.id:
126			raise ValueError('id for object %s was unexpected.' % model_object)
127			try:
128			model_object = cls._get_impl().insert(model_object)
129			except NotUniqueError as e:
130			if log_not_unique_error_as_debug:
131			LOG.debug('Conflict while trying to save in DB: %s.', str(e))
132			else:
133			LOG.exception('Conflict while trying to save in DB.')
134			# On a conflict determine the conflicting object and return its id in
135			# the raised exception.
136			conflict_object = cls._get_by_object(model_object)
137			conflict_id = str(conflict_object.id) if conflict_object else None
138			message = str(e)
139			raise StackStormDBObjectConflictError(message=message, conflict_id=conflict_id,
140			model_object=model_object)
141
142			# Publish internal event on the message bus
143			if publish:
144			try:
145			cls.publish_create(model_object)
146			except:
147			LOG.exception('Publish failed.')
148
149			# Dispatch trigger
150			if dispatch_trigger:
151			try:
152			cls.dispatch_create_trigger(model_object)
153			except:
154			LOG.exception('Trigger dispatch failed.')
155
156			return model_object
157
158			@classmethod
159			def add_or_update(cls, model_object, publish=True, dispatch_trigger=True,
160			log_not_unique_error_as_debug=False):
161			pre_persist_id = model_object.id
162			try:
163			model_object = cls._get_impl().add_or_update(model_object)
164			except NotUniqueError as e:
165			if log_not_unique_error_as_debug:
166			LOG.debug('Conflict while trying to save in DB: %s.', str(e))
167			else:
168			LOG.exception('Conflict while trying to save in DB.')
169			# On a conflict determine the conflicting object and return its id in
170			# the raised exception.
171			conflict_object = cls._get_by_object(model_object)
172			conflict_id = str(conflict_object.id) if conflict_object else None
173			message = str(e)
174			raise StackStormDBObjectConflictError(message=message, conflict_id=conflict_id,
175			model_object=model_object)
176
177			is_update = str(pre_persist_id) == str(model_object.id)
178
179			# Publish internal event on the message bus
180			if publish:
181			try:
182			if is_update:
183			cls.publish_update(model_object)
184			else:
185			cls.publish_create(model_object)
186			except:
187			LOG.exception('Publish failed.')
188
189			# Dispatch trigger
190			if dispatch_trigger:
191			try:
192			if is_update:
193			cls.dispatch_update_trigger(model_object)
194			else:
195			cls.dispatch_create_trigger(model_object)
196			except:
197			LOG.exception('Trigger dispatch failed.')
198
199			return model_object
200
201			@classmethod
202			def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs):
203			"""
204			Use this method when -
205			* upsert=False is desired
206			* special operators like push, push_all are to be used.
207			"""
208			cls._get_impl().update(model_object, **kwargs)
209			# update does not return the object but a flag; likely success/fail but docs
210			# are not very good on this one so ignoring. Explicitly get the object from
211			# DB abd return.
212			model_object = cls.get_by_id(model_object.id)
213
214			# Publish internal event on the message bus
215			if publish:
216			try:
217			cls.publish_update(model_object)
218			except:
219			LOG.exception('Publish failed.')
220
221			# Dispatch trigger
222			if dispatch_trigger:
223			try:
224			cls.dispatch_update_trigger(model_object)
225			except:
226			LOG.exception('Trigger dispatch failed.')
227
228			return model_object
229
230			@classmethod
231			def delete(cls, model_object, publish=True, dispatch_trigger=True):
232			persisted_object = cls._get_impl().delete(model_object)
233
234			# Publish internal event on the message bus
235			if publish:
236			try:
237			cls.publish_delete(model_object)
238			except Exception:
239			LOG.exception('Publish failed.')
240
241			# Dispatch trigger
242			if dispatch_trigger:
243			try:
244			cls.dispatch_delete_trigger(model_object)
245			except Exception:
246			LOG.exception('Trigger dispatch failed.')
247
248			return persisted_object
249
250			####################################################
251			# Internal event bus message publish related methods
252			####################################################
253
254			@classmethod
255			def publish_create(cls, model_object):
256			publisher = cls._get_publisher()
257			if publisher:
258			publisher.publish_create(model_object)
259
260			@classmethod
261			def publish_update(cls, model_object):
262			publisher = cls._get_publisher()
263			if publisher:
264			publisher.publish_update(model_object)
265
266			@classmethod
267			def publish_delete(cls, model_object):
268			publisher = cls._get_publisher()
269			if publisher:
270			publisher.publish_delete(model_object)
271
272			############################################
273			# Internal trigger dispatch related methods
274			###########################################
275
276			@classmethod
277			def dispatch_create_trigger(cls, model_object):
278			"""
279			Dispatch a resource-specific trigger which indicates a new resource has been created.
280			"""
281			return cls._dispatch_operation_trigger(operation='create', model_object=model_object)
282
283			@classmethod
284			def dispatch_update_trigger(cls, model_object):
285			"""
286			Dispatch a resource-specific trigger which indicates an existing resource has been updated.
287			"""
288			return cls._dispatch_operation_trigger(operation='update', model_object=model_object)
289
290			@classmethod
291			def dispatch_delete_trigger(cls, model_object):
292			"""
293			Dispatch a resource-specific trigger which indicates an existing resource has been
294			deleted.
295			"""
296			return cls._dispatch_operation_trigger(operation='delete', model_object=model_object)
297
298			@classmethod
299			def _get_trigger_ref_for_operation(cls, operation):
300			trigger_ref = cls.operation_to_trigger_ref_map.get(operation, None)
301
302			if not trigger_ref:
303			raise ValueError('Trigger ref not specified for operation: %s' % (operation))
304
305			return trigger_ref
306
307			@classmethod
308			def _dispatch_operation_trigger(cls, operation, model_object):
309			if operation not in cls.dispatch_trigger_for_operations:
310			return
311
312			trigger = cls._get_trigger_ref_for_operation(operation=operation)
313
314			object_payload = cls.api_model_cls.from_model(model_object, mask_secrets=True).__json__()
315			payload = {
316			'object': object_payload
317			}
318			return cls._dispatch_trigger(operation=operation, trigger=trigger, payload=payload)
319
320			@classmethod
321			def _dispatch_trigger(cls, operation, trigger, payload):
322			if operation not in cls.dispatch_trigger_for_operations:
323			return
324
325			dispatcher = cls._get_dispatcher()
326			return dispatcher.dispatch(trigger=trigger, payload=payload)
327
328
329			class ContentPackResource(Access):
330
331			@classmethod
332			def get_by_ref(cls, ref):
333			if not ref:
334			return None
335
336			ref_obj = ResourceReference.from_string_reference(ref=ref)
337			result = cls.query(name=ref_obj.name,
338			pack=ref_obj.pack).first()
339			return result
340
341			@classmethod
342			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...
343			# For an object with a resourcepack pack.name is unique.
344			name = getattr(object, 'name', '')
345			pack = getattr(object, 'pack', '')
346			return cls.get_by_ref(ResourceReference.to_string_reference(pack=pack, name=name))
347
348
349			class StatusBasedResource(Access):
350			"""Persistence layer for models that needs to publish status to the message queue."""
351
352			@classmethod
353			def publish_status(cls, model_object):
354			"""Publish the object status to the message queue.
355
356			Publish the instance of the model as payload with the status
357			as routing key to the message queue via the StatePublisher.
358
359			:param model_object: An instance of the model.
360			:type model_object: ``object``
361			"""
362			publisher = cls._get_publisher()
363			if publisher:
364			publisher.publish_state(model_object, getattr(model_object, 'status', None))
365

Plexxi / st2

GitHub Access Token became invalid

Push — develop-v1.6.0 ( 9d5181...7efb31 )

Access.add_or_update() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like