goblin.element.ElementMeta.__new__() - Code Metrics - goblin-ogm/goblin - Measure and Improve Code Quality continuously with Scrutinizer

goblin.element.ElementMeta.new() D
last analyzed 2022-05-13 18:52 UTC

↳ Parent: goblin.element

Complexity

Conditions

Size

Total Lines	36
Code Lines	35

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	35
nop	5
dl	0
loc	36
rs	4.8
c	0
b	0
f	0

How to fix Complexity

"""Module defining graph elements."""

import logging

import inflection # type: ignore
from gremlin_python.process.traversal import Cardinality # type: ignore
from enum import Enum

from goblin import abc, exception, mapper, properties
from abc import ABCMeta

logger = logging.getLogger(__name__)

class ImmutableMode(Enum):
    OFF = 0
    SIMPLE = 1

class LockingMode(Enum):
    OFF = 0
    OPTIMISTIC_LOCKING = 1

class ElementMeta(ABCMeta):
    """
    Metaclass for graph elements. Responsible for creating the
    :py:class:`Mapping<goblin.mapper.Mapping>` object and replacing user
    defined :py:class:`goblin.properties.Property` with
    :py:class:`goblin.properties.PropertyDescriptor`.
    """

    def __new__(cls, name, bases, namespace, **kwds):
        props = {}
        if name == 'VertexProperty':
            element_type = name.lower()
        elif bases:
            element_type = bases[0].__type__
            if element_type not in ['vertex', 'edge']:
                element_type = bases[0].__name__.lower()
            for base in bases:
                base_props = getattr(base, '__properties__', {})
                props.update(base_props)
        else:
            element_type = name.lower()
        namespace['__type__'] = element_type
        if not namespace.get('__label__', None):
            namespace['__label__'] = inflection.underscore(name)
        new_namespace = {}
        props.pop('id', None)
        for k, v in namespace.items():
            if isinstance(v, abc.BaseProperty):
                if element_type == 'edge' and hasattr(v, 'cardinality'):
                    raise exception.MappingError(
                        'Edge property cannot have set/list cardinality')
                props[k] = v
                if k != 'id':
                    if not v.db_name:
                        v.db_name = v.db_name_factory(k,
                                                      namespace['__label__'])
                v = v.__descriptor__(k, v)
            new_namespace[k] = v
        new_namespace['__mapping__'] = mapper.create_mapping(namespace, props)
        new_namespace['__properties__'] = props
        new_namespace['__immutable__'] = namespace.get('__immutable__', ImmutableMode.OFF)
        new_namespace['__locking__'] = namespace.get('__locking__', LockingMode.OFF)
        result = ABCMeta.__new__(cls, name, bases, new_namespace)
        return result


class Element(metaclass=ElementMeta):
    """Base class for classes that implement the Element property interface"""

    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            if not (hasattr(self, key) and isinstance(
                    getattr(self, key), properties.PropertyDescriptor)):
                raise AssertionError(
                    "No such property: {} for element {}".format(
                        key, self.__class__.__name__))
            setattr(self, key, value)

    id = properties.IdProperty(properties.Generic)
    dirty = properties.Property(properties.String)


class VertexPropertyDescriptor:
    """
    Descriptor that validates user property input and gets/sets properties
    as instance attributes.
    """

    def __init__(self, name, vertex_property):
        self._prop_name = name
        self._name = '_' + name
        self._vertex_property = vertex_property.__class__
        self._data_type = vertex_property.data_type
        self._default = vertex_property.default
        self._cardinality = vertex_property._cardinality

    def __get__(self, obj, objtype):
        if obj is None:
            return getattr(objtype.__mapping__, self._prop_name)
        default = self._default
        if default is not None:

            default = self._data_type.validate_vertex_prop(
                default, self._cardinality, self._vertex_property,
                self._data_type)
        return getattr(obj, self._name, default)

    def __set__(self, obj, val):
        if val is not None:
            val = self._data_type.validate_vertex_prop(
                val, self._cardinality, self._vertex_property, self._data_type)
        setattr(obj, self._name, val)


class VertexProperty(Element, abc.BaseProperty):
    """Base class for user defined vertex properties."""

    __descriptor__ = VertexPropertyDescriptor

    def __init__(self,
                 data_type,
                 *,
                 default=None,
                 db_name=None,
                 card=None,
                 db_name_factory=None):
        if not db_name_factory:
            def db_name_factory(x, y):
                pass
        if isinstance(data_type, type):
            data_type = data_type()
        self._db_name_factory = db_name_factory
        self._data_type = data_type
        self._default = default
        self._db_name = db_name
        self._val = None
        if card is None:
            card = Cardinality.single
        self._cardinality = card

    def to_dict(self):
        result = {
            '__label__': self.__label__,
            '__type__': self.__type__,
            '__value__': self._val
        }
        for key, value in self.__properties__.items():
            prop = getattr(self, key, None)
            result[key] = prop
        return result

    def from_dict(self, d):
        d.pop('__label__')
        d.pop('__type__')
        d.pop('__value__')
        for key, value in d.items():
            setattr(self, key, value)

    @property
    def default(self):
        return self._default

    @property
    def data_type(self):
        return self._data_type

    @property
    def db_name_factory(self):
        return self._db_name_factory

    def getvalue(self):
        return self._val

    def setvalue(self, val):
        self._val = val

    value = property(getvalue, setvalue)

    def getdb_name(self):
        return self._db_name

    def setgetdb_name(self, val):
        self._db_name = val

    db_name = property(getdb_name, setgetdb_name)

    @property
    def cardinality(self):
        return self._cardinality

    def __repr__(self):
        return '<{}(type={}, value={})'.format(self.__class__.__name__,
                                               self._data_type, self.value)


class Vertex(Element):
    """Base class for user defined Vertex classes"""

    def to_dict(self):
        result = {'__label__': self.__label__, '__type__': self.__type__}
        for key, value in self.__properties__.items():
            vert_prop = getattr(self, key, None)
            if isinstance(vert_prop, (list, set)):
                vert_prop = [vp.to_dict() for vp in vert_prop]
            elif isinstance(vert_prop, VertexProperty):
                vert_prop = vert_prop.to_dict()
            result[key] = vert_prop
        return result

    @classmethod
    def from_dict(cls, d):
        elem = cls()
        d.pop('__label__')
        d.pop('__type__')
        for key, value in d.items():
            if isinstance(value, list):
                first_prop = value[0]
                setattr(elem, key, first_prop['__value__'])
                if isinstance(getattr(elem, key), list):
                    getattr(elem, key)[0].from_dict(first_prop)
                    for prop in value[1:]:
                        getattr(elem, key).append(prop['__value__'])
                        getattr(elem, key)[-1].from_dict(prop)

                elif isinstance(getattr(elem, key), set):
                    getattr(elem,
                            key)(first_prop['__value__']).from_dict(first_prop)
                    for prop in value[1:]:
                        val = prop['__value__']
                        getattr(elem, key).add(val)
                        getattr(elem, key)(val).from_dict(prop)
                else:
                    raise Exception("not a list or set property")
            elif isinstance(value, dict):
                setattr(elem, key, value['__value__'])
                getattr(elem, key).from_dict(value)
            else:
                setattr(elem, key, value)
        return elem


class GenericVertex(Vertex):
    """
    Class used to build vertices when user defined vertex class is not
    available. Generally not instantiated by end user.
    """
    pass


class Edge(Element):
    """
    Base class for user defined Edge classes.

    :param Vertex source: Source (outV) vertex
    :param Vertex target: Target (inV) vertex
    """

    def __init__(self, source=None, target=None):
        self.source = source
        self.target = target

    def to_dict(self, source=None, target=None):
        if not source:
            source = self.source.to_dict()
        if not target:
            target = self.target.to_dict()
        result = {
            '__label__': self.__label__,
            '__type__': self.__type__,
            'source': source,
            'target': target
        }
        for key, value in self.__properties__.items():
            prop = getattr(self, key, None)
            result[key] = prop
        return result

    @classmethod
    def from_dict(cls, d):
        elem = cls()
        d.pop('__label__')
        d.pop('__type__')
        for key, value in d.items():
            setattr(elem, key, value)
        return elem

    def getsource(self):
        return self._source

    def setsource(self, vertex):
        assert isinstance(vertex, Vertex) or vertex is None
        self._source = vertex

    def delsource(self):
        del self._source

    source = property(getsource, setsource, delsource)

    def gettarget(self):
        return self._target

    def settarget(self, vertex):
        assert isinstance(vertex, Vertex) or vertex is None
        self._target = vertex

    def deltarget(self):
        del self._target

    target = property(gettarget, settarget, deltarget)


class GenericEdge(Edge):
    """
    Class used to build edges when user defined edges class is not available.
    Generally not instantiated by end user.
    """
    pass


1			"""Module defining graph elements."""
2
3			import logging
4
5			import inflection # type: ignore
6			from gremlin_python.process.traversal import Cardinality # type: ignore
7			from enum import Enum
8
9			from goblin import abc, exception, mapper, properties
10			from abc import ABCMeta
11
12			logger = logging.getLogger(__name__)
13
14			class ImmutableMode(Enum):
15			OFF = 0
16			SIMPLE = 1
17
18			class LockingMode(Enum):
19			OFF = 0
20			OPTIMISTIC_LOCKING = 1
21
22			class ElementMeta(ABCMeta):
23			"""
24			Metaclass for graph elements. Responsible for creating the
25			:py:class:`Mapping<goblin.mapper.Mapping>` object and replacing user
26			defined :py:class:`goblin.properties.Property` with
27			:py:class:`goblin.properties.PropertyDescriptor`.
28			"""
29
30			def __new__(cls, name, bases, namespace, **kwds):
31			props = {}
32			if name == 'VertexProperty':
33			element_type = name.lower()
34			elif bases:
35			element_type = bases[0].__type__
36			if element_type not in ['vertex', 'edge']:
37			element_type = bases[0].__name__.lower()
38			for base in bases:
39			base_props = getattr(base, '__properties__', {})
40			props.update(base_props)
41			else:
42			element_type = name.lower()
43			namespace['__type__'] = element_type
44			if not namespace.get('__label__', None):
45			namespace['__label__'] = inflection.underscore(name)
46			new_namespace = {}
47			props.pop('id', None)
48			for k, v in namespace.items():
49			if isinstance(v, abc.BaseProperty):
50			if element_type == 'edge' and hasattr(v, 'cardinality'):
51			raise exception.MappingError(
52			'Edge property cannot have set/list cardinality')
53			props[k] = v
54			if k != 'id':
55			if not v.db_name:
56			v.db_name = v.db_name_factory(k,
57			namespace['__label__'])
58			v = v.__descriptor__(k, v)
59			new_namespace[k] = v
60			new_namespace['__mapping__'] = mapper.create_mapping(namespace, props)
61			new_namespace['__properties__'] = props
62			new_namespace['__immutable__'] = namespace.get('__immutable__', ImmutableMode.OFF)
63			new_namespace['__locking__'] = namespace.get('__locking__', LockingMode.OFF)
64			result = ABCMeta.__new__(cls, name, bases, new_namespace)
65			return result
66
67
68			class Element(metaclass=ElementMeta):
69			"""Base class for classes that implement the Element property interface"""
70
71			def __init__(self, **kwargs):
72			for key, value in kwargs.items():
73			if not (hasattr(self, key) and isinstance(
74			getattr(self, key), properties.PropertyDescriptor)):
75			raise AssertionError(
76			"No such property: {} for element {}".format(
77			key, self.__class__.__name__))
78			setattr(self, key, value)
79
80			id = properties.IdProperty(properties.Generic)
81			dirty = properties.Property(properties.String)
82
83
84			class VertexPropertyDescriptor:
85			"""
86			Descriptor that validates user property input and gets/sets properties
87			as instance attributes.
88			"""
89
90			def __init__(self, name, vertex_property):
91			self._prop_name = name
92			self._name = '_' + name
93			self._vertex_property = vertex_property.__class__
94			self._data_type = vertex_property.data_type
95			self._default = vertex_property.default
96			self._cardinality = vertex_property._cardinality
97
98			def __get__(self, obj, objtype):
99			if obj is None:
100			return getattr(objtype.__mapping__, self._prop_name)
101			default = self._default
102			if default is not None:
103
104			default = self._data_type.validate_vertex_prop(
105			default, self._cardinality, self._vertex_property,
106			self._data_type)
107			return getattr(obj, self._name, default)
108
109			def __set__(self, obj, val):
110			if val is not None:
111			val = self._data_type.validate_vertex_prop(
112			val, self._cardinality, self._vertex_property, self._data_type)
113			setattr(obj, self._name, val)
114
115
116			class VertexProperty(Element, abc.BaseProperty):
117			"""Base class for user defined vertex properties."""
118
119			__descriptor__ = VertexPropertyDescriptor
120
121			def __init__(self,
122			data_type,
123			*,
124			default=None,
125			db_name=None,
126			card=None,
127			db_name_factory=None):
128			if not db_name_factory:
129			def db_name_factory(x, y):
130			pass
131			if isinstance(data_type, type):
132			data_type = data_type()
133			self._db_name_factory = db_name_factory
134			self._data_type = data_type
135			self._default = default
136			self._db_name = db_name
137			self._val = None
138			if card is None:
139			card = Cardinality.single
140			self._cardinality = card
141
142			def to_dict(self):
143			result = {
144			'__label__': self.__label__,
145			'__type__': self.__type__,
146			'__value__': self._val
147			}
148			for key, value in self.__properties__.items():
149			prop = getattr(self, key, None)
150			result[key] = prop
151			return result
152
153			def from_dict(self, d):
154			d.pop('__label__')
155			d.pop('__type__')
156			d.pop('__value__')
157			for key, value in d.items():
158			setattr(self, key, value)
159
160			@property
161			def default(self):
162			return self._default
163
164			@property
165			def data_type(self):
166			return self._data_type
167
168			@property
169			def db_name_factory(self):
170			return self._db_name_factory
171
172			def getvalue(self):
173			return self._val
174
175			def setvalue(self, val):
176			self._val = val
177
178			value = property(getvalue, setvalue)
179
180			def getdb_name(self):
181			return self._db_name
182
183			def setgetdb_name(self, val):
184			self._db_name = val
185
186			db_name = property(getdb_name, setgetdb_name)
187
188			@property
189			def cardinality(self):
190			return self._cardinality
191
192			def __repr__(self):
193			return '<{}(type={}, value={})'.format(self.__class__.__name__,
194			self._data_type, self.value)
195
196
197			class Vertex(Element):
198			"""Base class for user defined Vertex classes"""
199
200			def to_dict(self):
201			result = {'__label__': self.__label__, '__type__': self.__type__}
202			for key, value in self.__properties__.items():
203			vert_prop = getattr(self, key, None)
204			if isinstance(vert_prop, (list, set)):
205			vert_prop = [vp.to_dict() for vp in vert_prop]
206			elif isinstance(vert_prop, VertexProperty):
207			vert_prop = vert_prop.to_dict()
208			result[key] = vert_prop
209			return result
210
211			@classmethod
212			def from_dict(cls, d):
213			elem = cls()
214			d.pop('__label__')
215			d.pop('__type__')
216			for key, value in d.items():
217			if isinstance(value, list):
218			first_prop = value[0]
219			setattr(elem, key, first_prop['__value__'])
220			if isinstance(getattr(elem, key), list):
221			getattr(elem, key)[0].from_dict(first_prop)
222			for prop in value[1:]:
223			getattr(elem, key).append(prop['__value__'])
224			getattr(elem, key)[-1].from_dict(prop)
225
226			elif isinstance(getattr(elem, key), set):
227			getattr(elem,
228			key)(first_prop['__value__']).from_dict(first_prop)
229			for prop in value[1:]:
230			val = prop['__value__']
231			getattr(elem, key).add(val)
232			getattr(elem, key)(val).from_dict(prop)
233			else:
234			raise Exception("not a list or set property")
235			elif isinstance(value, dict):
236			setattr(elem, key, value['__value__'])
237			getattr(elem, key).from_dict(value)
238			else:
239			setattr(elem, key, value)
240			return elem
241
242
243			class GenericVertex(Vertex):
244			"""
245			Class used to build vertices when user defined vertex class is not
246			available. Generally not instantiated by end user.
247			"""
248			pass
249
250
251			class Edge(Element):
252			"""
253			Base class for user defined Edge classes.
254
255			:param Vertex source: Source (outV) vertex
256			:param Vertex target: Target (inV) vertex
257			"""
258
259			def __init__(self, source=None, target=None):
260			self.source = source
261			self.target = target
262
263			def to_dict(self, source=None, target=None):
264			if not source:
265			source = self.source.to_dict()
266			if not target:
267			target = self.target.to_dict()
268			result = {
269			'__label__': self.__label__,
270			'__type__': self.__type__,
271			'source': source,
272			'target': target
273			}
274			for key, value in self.__properties__.items():
275			prop = getattr(self, key, None)
276			result[key] = prop
277			return result
278
279			@classmethod
280			def from_dict(cls, d):
281			elem = cls()
282			d.pop('__label__')
283			d.pop('__type__')
284			for key, value in d.items():
285			setattr(elem, key, value)
286			return elem
287
288			def getsource(self):
289			return self._source
290
291			def setsource(self, vertex):
292			assert isinstance(vertex, Vertex) or vertex is None
293			self._source = vertex
294
295			def delsource(self):
296			del self._source
297
298			source = property(getsource, setsource, delsource)
299
300			def gettarget(self):
301			return self._target
302
303			def settarget(self, vertex):
304			assert isinstance(vertex, Vertex) or vertex is None
305			self._target = vertex
306
307			def deltarget(self):
308			del self._target
309
310			target = property(gettarget, settarget, deltarget)
311
312
313			class GenericEdge(Edge):
314			"""
315			Class used to build edges when user defined edges class is not available.
316			Generally not instantiated by end user.
317			"""
318			pass
319

goblin-ogm / goblin

goblin.element.ElementMeta.__new__() D last analyzed 2022-05-13 18:52 UTC

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

goblin.element.ElementMeta.new() D
last analyzed 2022-05-13 18:52 UTC