UIDFieldMixin.has_valid_uid() - Code Metrics - Inspection of "Merge pull request #3864 from StackStorm/fix_rule_..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2df409...505678 )

unknown

created 2017-11-27 11:00 UTC

UIDFieldMixin.has_valid_uid() A

↳ Parent: UIDFieldMixin

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	8
rs	9.4285
c	0
b	0
f	0

# 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 datetime

import bson
import six
import mongoengine as me
from oslo_config import cfg

from st2common.util import mongoescape
from st2common.models.base import DictSerializableClassMixin
from st2common.models.system.common import ResourceReference
from st2common.constants.types import ResourceType

__all__ = [
    'StormFoundationDB',
    'StormBaseDB',

    'EscapedDictField',
    'EscapedDynamicField',
    'TagField',

    'RefFieldMixin',
    'UIDFieldMixin',
    'TagsMixin',
    'ContentPackResourceMixin'
]

JSON_UNFRIENDLY_TYPES = (datetime.datetime, bson.ObjectId, me.EmbeddedDocument)


class StormFoundationDB(me.Document, DictSerializableClassMixin):
    """
    Base abstraction for a model entity. This foundation class should only be directly
    inherited from the application domain models.
    """

    # Variable representing a type of this resource
    RESOURCE_TYPE = ResourceType.UNKNOWN

    # We explicitly assign the manager so pylint know what type objects is
    objects = me.queryset.QuerySetManager()

    # Note: In mongoengine >= 0.10 "id" field is automatically declared on all
    # the documents and declaring it ourselves causes a lot of issues so we
    # don't do that

    # see http://docs.mongoengine.org/guide/defining-documents.html#abstract-classes
    meta = {
        'abstract': True
    }

    def __str__(self):
        attrs = list()
        for k in sorted(self._fields.keys()):
            v = getattr(self, k)
            v = '"%s"' % str(v) if type(v) in [str, unicode, datetime.datetime] else str(v)

            attrs.append('%s=%s' % (k, v))
        return '%s(%s)' % (self.__class__.__name__, ', '.join(attrs))

    def get_resource_type(self):
        return self.RESOURCE_TYPE

    def mask_secrets(self, value):
        """
        Process the model dictionary and mask secret values.

        :type value: ``dict``
        :param value: Document dictionary.

        :rtype: ``dict``
        """
        return value

    def to_serializable_dict(self, mask_secrets=False):
        """
        Serialize database model to a dictionary.

        :param mask_secrets: True to mask secrets in the resulting dict.
        :type mask_secrets: ``boolean``

        :rtype: ``dict``
        """
        serializable_dict = {}
        for k in sorted(six.iterkeys(self._fields)):
            v = getattr(self, k)
            v = str(v) if isinstance(v, JSON_UNFRIENDLY_TYPES) else v
            serializable_dict[k] = v

        if mask_secrets and cfg.CONF.log.mask_secrets:
            serializable_dict = self.mask_secrets(value=serializable_dict)

        return serializable_dict


class StormBaseDB(StormFoundationDB):
    """Abstraction for a user content model."""

    name = me.StringField(required=True, unique=True)
    description = me.StringField()

    # see http://docs.mongoengine.org/guide/defining-documents.html#abstract-classes
    meta = {
        'abstract': True
    }


class EscapedDictField(me.DictField):

    def to_mongo(self, value, use_db_field=True, fields=None):
        value = mongoescape.escape_chars(value)
        return super(EscapedDictField, self).to_mongo(value=value, use_db_field=use_db_field,
                                                      fields=fields)

    def to_python(self, value):
        value = super(EscapedDictField, self).to_python(value)
        return mongoescape.unescape_chars(value)

    def validate(self, value):
        if not isinstance(value, dict):
            self.error('Only dictionaries may be used in a DictField')
        if me.fields.key_not_string(value):
            self.error("Invalid dictionary key - documents must have only string keys")
        me.base.ComplexBaseField.validate(self, value)


class EscapedDynamicField(me.DynamicField):

    def to_mongo(self, value, use_db_field=True, fields=None):
        value = mongoescape.escape_chars(value)
        return super(EscapedDynamicField, self).to_mongo(value=value, use_db_field=use_db_field,
                                                         fields=fields)

    def to_python(self, value):
        value = super(EscapedDynamicField, self).to_python(value)
        return mongoescape.unescape_chars(value)


class TagField(me.EmbeddedDocument):
    """
    To be attached to a db model object for the purpose of providing supplemental
    information.
    """
    name = me.StringField(max_length=1024)
    value = me.StringField(max_length=1024)


class TagsMixin(object):
    """
    Mixin to include tags on an object.
    """
    tags = me.ListField(field=me.EmbeddedDocumentField(TagField))

    @classmethod
    def get_indices(cls):
        return ['tags.name', 'tags.value']


class RefFieldMixin(object):
    """
    Mixin class which adds "ref" field to the class inheriting from it.
    """

    ref = me.StringField(required=True, unique=True)


class UIDFieldMixin(object):
    """
    Mixin class which adds "uid" field to the class inheriting from it.

    UID field is a unique identifier which we can be used to unambiguously reference a resource in
    the system.
    """

    UID_SEPARATOR = ':'  # TODO: Move to constants

    RESOURCE_TYPE = abc.abstractproperty
    UID_FIELDS = abc.abstractproperty

    uid = me.StringField(required=True)

    @classmethod
    def get_indexes(cls):
        # Note: We use a special sparse index so we don't need to pre-populate "uid" for existing
        # models in the database before ensure_indexes() is called.
        # This field gets populated in the constructor which means it will be lazily assigned next
        # time the model is saved (e.g. once register-content is ran).
        indexes = [
            {
                'fields': ['uid'],
                'unique': True,
                'sparse': True
            }
        ]
        return indexes

    def get_uid(self):
        """
        Return an object UID constructed from the object properties / fields.

        :rtype: ``str``
        """
        parts = []
        parts.append(self.RESOURCE_TYPE)

        for field in self.UID_FIELDS:
            value = getattr(self, field, None) or ''
            parts.append(value)

        uid = self.UID_SEPARATOR.join(parts)
        return uid

    def get_uid_parts(self):
        """
        Return values for fields which make up the UID.

        :rtype: ``list``
        """
        parts = self.uid.split(self.UID_SEPARATOR)  # pylint: disable=no-member
        parts = [part for part in parts if part.strip()]
        return parts

    def has_valid_uid(self):
        """
        Return True if object contains a valid id (aka all parts contain a valid value).

        :rtype: ``bool``
        """
        parts = self.get_uid_parts()
        return len(parts) == len(self.UID_FIELDS) + 1


class ContentPackResourceMixin(object):
    """
    Mixin class provides utility methods for models which belong to a pack.
    """

    def get_pack_uid(self):
        """
        Return an UID of a pack this resource belongs to.

        :rtype ``str``
        """
        parts = [ResourceType.PACK, self.pack]
        uid = UIDFieldMixin.UID_SEPARATOR.join(parts)
        return uid

    def get_reference(self):
        """
        Retrieve referene object for this model.

        :rtype: :class:`ResourceReference`
        """
        if getattr(self, 'ref', None):
            ref = ResourceReference.from_string_reference(ref=self.ref)
        else:
            ref = ResourceReference(pack=self.pack, name=self.name)

        return ref


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			import datetime
18
19			import bson
20			import six
21			import mongoengine as me
22			from oslo_config import cfg
23
24			from st2common.util import mongoescape
25			from st2common.models.base import DictSerializableClassMixin
26			from st2common.models.system.common import ResourceReference
27			from st2common.constants.types import ResourceType
28
29			__all__ = [
30			'StormFoundationDB',
31			'StormBaseDB',
32
33			'EscapedDictField',
34			'EscapedDynamicField',
35			'TagField',
36
37			'RefFieldMixin',
38			'UIDFieldMixin',
39			'TagsMixin',
40			'ContentPackResourceMixin'
41			]
42
43			JSON_UNFRIENDLY_TYPES = (datetime.datetime, bson.ObjectId, me.EmbeddedDocument)
44
45
46			class StormFoundationDB(me.Document, DictSerializableClassMixin):
47			"""
48			Base abstraction for a model entity. This foundation class should only be directly
49			inherited from the application domain models.
50			"""
51
52			# Variable representing a type of this resource
53			RESOURCE_TYPE = ResourceType.UNKNOWN
54
55			# We explicitly assign the manager so pylint know what type objects is
56			objects = me.queryset.QuerySetManager()
57
58			# Note: In mongoengine >= 0.10 "id" field is automatically declared on all
59			# the documents and declaring it ourselves causes a lot of issues so we
60			# don't do that
61
62			# see http://docs.mongoengine.org/guide/defining-documents.html#abstract-classes
63			meta = {
64			'abstract': True
65			}
66
67			def __str__(self):
68			attrs = list()
69			for k in sorted(self._fields.keys()):
70			v = getattr(self, k)
71			v = '"%s"' % str(v) if type(v) in [str, unicode, datetime.datetime] else str(v)
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report Undefined variable 'unicode' Loading history...
72			attrs.append('%s=%s' % (k, v))
73			return '%s(%s)' % (self.__class__.__name__, ', '.join(attrs))
74
75			def get_resource_type(self):
76			return self.RESOURCE_TYPE
77
78			def mask_secrets(self, value):
79			"""
80			Process the model dictionary and mask secret values.
81
82			:type value: ``dict``
83			:param value: Document dictionary.
84
85			:rtype: ``dict``
86			"""
87			return value
88
89			def to_serializable_dict(self, mask_secrets=False):
90			"""
91			Serialize database model to a dictionary.
92
93			:param mask_secrets: True to mask secrets in the resulting dict.
94			:type mask_secrets: ``boolean``
95
96			:rtype: ``dict``
97			"""
98			serializable_dict = {}
99			for k in sorted(six.iterkeys(self._fields)):
100			v = getattr(self, k)
101			v = str(v) if isinstance(v, JSON_UNFRIENDLY_TYPES) else v
102			serializable_dict[k] = v
103
104			if mask_secrets and cfg.CONF.log.mask_secrets:
105			serializable_dict = self.mask_secrets(value=serializable_dict)
106
107			return serializable_dict
108
109
110			class StormBaseDB(StormFoundationDB):
111			"""Abstraction for a user content model."""
112
113			name = me.StringField(required=True, unique=True)
114			description = me.StringField()
115
116			# see http://docs.mongoengine.org/guide/defining-documents.html#abstract-classes
117			meta = {
118			'abstract': True
119			}
120
121
122			class EscapedDictField(me.DictField):
123
124			def to_mongo(self, value, use_db_field=True, fields=None):
125			value = mongoescape.escape_chars(value)
126			return super(EscapedDictField, self).to_mongo(value=value, use_db_field=use_db_field,
127			fields=fields)
128
129			def to_python(self, value):
130			value = super(EscapedDictField, self).to_python(value)
131			return mongoescape.unescape_chars(value)
132
133			def validate(self, value):
134			if not isinstance(value, dict):
135			self.error('Only dictionaries may be used in a DictField')
136			if me.fields.key_not_string(value):
137			self.error("Invalid dictionary key - documents must have only string keys")
138			me.base.ComplexBaseField.validate(self, value)
139
140
141			class EscapedDynamicField(me.DynamicField):
142
143			def to_mongo(self, value, use_db_field=True, fields=None):
144			value = mongoescape.escape_chars(value)
145			return super(EscapedDynamicField, self).to_mongo(value=value, use_db_field=use_db_field,
146			fields=fields)
147
148			def to_python(self, value):
149			value = super(EscapedDynamicField, self).to_python(value)
150			return mongoescape.unescape_chars(value)
151
152
153			class TagField(me.EmbeddedDocument):
154			"""
155			To be attached to a db model object for the purpose of providing supplemental
156			information.
157			"""
158			name = me.StringField(max_length=1024)
159			value = me.StringField(max_length=1024)
160
161
162			class TagsMixin(object):
163			"""
164			Mixin to include tags on an object.
165			"""
166			tags = me.ListField(field=me.EmbeddedDocumentField(TagField))
167
168			@classmethod
169			def get_indices(cls):
170			return ['tags.name', 'tags.value']
171
172
173			class RefFieldMixin(object):
174			"""
175			Mixin class which adds "ref" field to the class inheriting from it.
176			"""
177
178			ref = me.StringField(required=True, unique=True)
179
180
181			class UIDFieldMixin(object):
182			"""
183			Mixin class which adds "uid" field to the class inheriting from it.
184
185			UID field is a unique identifier which we can be used to unambiguously reference a resource in
186			the system.
187			"""
188
189			UID_SEPARATOR = ':' # TODO: Move to constants
190
191			RESOURCE_TYPE = abc.abstractproperty
192			UID_FIELDS = abc.abstractproperty
193
194			uid = me.StringField(required=True)
195
196			@classmethod
197			def get_indexes(cls):
198			# Note: We use a special sparse index so we don't need to pre-populate "uid" for existing
199			# models in the database before ensure_indexes() is called.
200			# This field gets populated in the constructor which means it will be lazily assigned next
201			# time the model is saved (e.g. once register-content is ran).
202			indexes = [
203			{
204			'fields': ['uid'],
205			'unique': True,
206			'sparse': True
207			}
208			]
209			return indexes
210
211			def get_uid(self):
212			"""
213			Return an object UID constructed from the object properties / fields.
214
215			:rtype: ``str``
216			"""
217			parts = []
218			parts.append(self.RESOURCE_TYPE)
219
220			for field in self.UID_FIELDS:
221			value = getattr(self, field, None) or ''
222			parts.append(value)
223
224			uid = self.UID_SEPARATOR.join(parts)
225			return uid
226
227			def get_uid_parts(self):
228			"""
229			Return values for fields which make up the UID.
230
231			:rtype: ``list``
232			"""
233			parts = self.uid.split(self.UID_SEPARATOR) # pylint: disable=no-member
234			parts = [part for part in parts if part.strip()]
235			return parts
236
237			def has_valid_uid(self):
238			"""
239			Return True if object contains a valid id (aka all parts contain a valid value).
240
241			:rtype: ``bool``
242			"""
243			parts = self.get_uid_parts()
244			return len(parts) == len(self.UID_FIELDS) + 1
245
246
247			class ContentPackResourceMixin(object):
248			"""
249			Mixin class provides utility methods for models which belong to a pack.
250			"""
251
252			def get_pack_uid(self):
253			"""
254			Return an UID of a pack this resource belongs to.
255
256			:rtype ``str``
257			"""
258			parts = [ResourceType.PACK, self.pack]
259			uid = UIDFieldMixin.UID_SEPARATOR.join(parts)
260			return uid
261
262			def get_reference(self):
263			"""
264			Retrieve referene object for this model.
265
266			:rtype: :class:`ResourceReference`
267			"""
268			if getattr(self, 'ref', None):
269			ref = ResourceReference.from_string_reference(ref=self.ref)
270			else:
271			ref = ResourceReference(pack=self.pack, name=self.name)
272
273			return ref
274

StackStorm / st2

Push — master ( 2df409...505678 )

UIDFieldMixin.has_valid_uid() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like