goblin.mapper.map_props_to_db() - Code Metrics - goblin-ogm/goblin - Measure and Improve Code Quality continuously with Scrutinizer

goblin.mapper.map_props_to_db() B
last analyzed 2022-05-13 18:52 UTC

↳ Parent: goblin.mapper

Complexity

Conditions

Size

Total Lines	22
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	19
nop	2
dl	0
loc	22
rs	8.5166
c	0
b	0
f	0

"""Helper functions and class to map between OGM Elements <-> DB Elements"""

from typing import Any, Dict
import functools
import logging

from goblin import exception
from gremlin_python.process.traversal import T

logger = logging.getLogger(__name__)


def map_props_to_db(element, mapping):
    """Convert OGM property names/values to DB property names/values"""
    property_tuples = []
    props = mapping.ogm_properties
    for ogm_name, (db_name, data_type) in props.items():
        val = getattr(element, ogm_name, None)
        if val and isinstance(val, (list, set)):
            card = None
            for v in val:
                metaprops = get_metaprops(v, v.__mapping__)
                property_tuples.append((card, db_name, data_type.to_db(
                    v.value), metaprops))
                card = v.cardinality
        else:
            if hasattr(val, '__mapping__'):
                metaprops = get_metaprops(val, val.__mapping__)
                val = val.value
            else:
                metaprops = None
            property_tuples.append((None, db_name, data_type.to_db(val),
                                    metaprops))
    return property_tuples


def get_metaprops(vertex_property, mapping):
    props = mapping.ogm_properties
    metaprops = {}
    for ogm_name, (db_name, data_type) in props.items():
        val = getattr(vertex_property, ogm_name, None)
        metaprops[db_name] = data_type.to_db(val)
    return metaprops


def map_vertex_to_ogm(result, props, element, *, mapping=None):
    """Map a vertex returned by DB to OGM vertex"""
    props.pop('id')
    label = props.pop('label')
    for db_name, value in props.items():
        metaprops = []
        if len(value) > 1:
            values = []
            for v in value:
                if isinstance(v, dict):
                    val = v.pop('value')
                    v.pop('key')
                    vid = v.pop('id')
                    if v:
                        v['id'] = vid
                        metaprops.append((val, v))
                    values.append(val)
                else:
                    values.append(v)
            value = values
        else:
            value = value[0]
            if isinstance(value, dict):
                val = value.pop('value')
                value.pop('key')
                vid = value.pop('id')
                if value:
                    value['id'] = vid
                    metaprops.append((val, value))
                value = val
        name, data_type = mapping.db_properties.get(db_name, (db_name, None))
        if data_type:
            value = data_type.to_ogm(value)
        setattr(element, name, value)
        if metaprops:
            vert_prop = getattr(element, name)
            if hasattr(vert_prop, 'mapper_func'):
                # Temporary hack for managers
                vert_prop.mapper_func(metaprops, vert_prop)
            else:
                vert_prop.__mapping__.mapper_func(metaprops, vert_prop)
    setattr(element, '__label__', label)
    setattr(element, 'id', result.id)
    return element


# TODO: temp hack
def get_hashable_id(val: Dict[str, Any]) -> Any:
    if isinstance(val, dict) and "@type" in val and "@value" in val:
        if val["@type"] == "janusgraph:RelationIdentifier":
            val = val["@value"].get("value", val["@value"]["relationId"])
    return val



def map_vertex_property_to_ogm(result, element, *, mapping=None):
    """Map a vertex returned by DB to OGM vertex"""
    for (val, metaprops) in result:
        if isinstance(element, list):
            current = element.vp_map.get(get_hashable_id(metaprops['id']))

            # This whole system needs to be reevaluated
            if not current:
                current = element(val)
                if isinstance(current, list):
                    for vp in current:
                        if not hasattr(vp, '_id'):
                            element.vp_map[get_hashable_id(
                                metaprops['id'])] = vp
                            current = vp
                            break
        elif isinstance(element, set):
            current = element(val)
        else:
            current = element
        for db_name, value in metaprops.items():
            name, data_type = mapping.db_properties.get(
                db_name, (db_name, None))
            if data_type:
                value = data_type.to_ogm(value)
            setattr(current, name, value)


def map_edge_to_ogm(result, props, element, *, mapping=None):
    """Map an edge returned by DB to OGM edge"""
    props.pop(T.id)
    label = props.pop(T.label)
    for db_name, value in props.items():
        name, data_type = mapping.db_properties.get(db_name, (db_name, None))
        if data_type:
            value = data_type.to_ogm(value)
        setattr(element, name, value)
    setattr(element, '__label__', label)
    setattr(element, 'id', result.id)
    # Currently not included in graphson
    # setattr(element.source, '__label__', result.outV.label)
    # setattr(element.target, '__label__', result.inV.label)
    sid = result.outV.id
    esid = getattr(element.source, 'id', None)
    if _check_id(sid, esid):
        from goblin.element import GenericVertex
        element.source = GenericVertex()
    tid = result.inV.id
    etid = getattr(element.target, 'id', None)
    if _check_id(tid, etid):
        from goblin.element import GenericVertex
        element.target = GenericVertex()
    setattr(element.source, 'id', sid)
    setattr(element.target, 'id', tid)
    return element


def _check_id(rid, eid):
    if eid and rid != eid:
        logger.warning('Edge vertex id has changed')
        return True
    return False


# DB <-> OGM Mapping
def create_mapping(namespace, properties):
    """Constructor for :py:class:`Mapping`"""
    element_type = namespace['__type__']
    if element_type == 'vertex':
        mapping_func = map_vertex_to_ogm
        mapping = Mapping(namespace, element_type, mapping_func, properties)
    elif element_type == 'edge':
        mapping_func = map_edge_to_ogm
        mapping = Mapping(namespace, element_type, mapping_func, properties)
    elif element_type == 'vertexproperty':
        mapping_func = map_vertex_property_to_ogm
        mapping = Mapping(namespace, element_type, mapping_func, properties)
    else:
        mapping = None
    return mapping


class Mapping:
    """
    This class stores the information necessary to map between an OGM element
    and a DB element.
    """

    def __init__(self, namespace, element_type, mapper_func, properties):
        self._label = namespace['__label__']
        self._element_type = element_type
        self._mapper_func = functools.partial(mapper_func, mapping=self)
        self._db_properties = {}
        self._ogm_properties = {}
        self._map_properties(properties)

    @property
    def label(self):
        """Element label"""
        return self._label

    @property
    def mapper_func(self):
        """Function responsible for mapping db results to ogm"""
        return self._mapper_func

    @property
    def db_properties(self):
        """A dictionary of property mappings"""
        return self._db_properties

    @property
    def ogm_properties(self):
        """A dictionary of property mappings"""
        return self._ogm_properties

    def __getattr__(self, value):
        try:
            mapping, _ = self._ogm_properties[value]
            return mapping
        except KeyError:
            raise exception.MappingError(
                "unrecognized property {} for class: {}".format(
                    value, self._element_type))

    def _map_properties(self, properties):
        for name, prop in properties.items():
            data_type = prop.data_type
            if prop.db_name:
                db_name = prop.db_name
            else:
                db_name = name
            if hasattr(prop, '__mapping__'):
                if not self._element_type == 'vertex':
                    raise exception.MappingError(
                        'Only vertices can have vertex properties')
            self._db_properties[db_name] = (name, data_type)
            self._ogm_properties[name] = (db_name, data_type)

    def __repr__(self):
        return '<{}(type={}, label={}, properties={})>'.format(
            self.__class__.__name__, self._element_type, self._label,
            self._ogm_properties)


1			"""Helper functions and class to map between OGM Elements <-> DB Elements"""
2
3			from typing import Any, Dict
4			import functools
5			import logging
6
7			from goblin import exception
8			from gremlin_python.process.traversal import T
9
10			logger = logging.getLogger(__name__)
11
12
13			def map_props_to_db(element, mapping):
14			"""Convert OGM property names/values to DB property names/values"""
15			property_tuples = []
16			props = mapping.ogm_properties
17			for ogm_name, (db_name, data_type) in props.items():
18			val = getattr(element, ogm_name, None)
19			if val and isinstance(val, (list, set)):
20			card = None
21			for v in val:
22			metaprops = get_metaprops(v, v.__mapping__)
23			property_tuples.append((card, db_name, data_type.to_db(
24			v.value), metaprops))
25			card = v.cardinality
26			else:
27			if hasattr(val, '__mapping__'):
28			metaprops = get_metaprops(val, val.__mapping__)
29			val = val.value
30			else:
31			metaprops = None
32			property_tuples.append((None, db_name, data_type.to_db(val),
33			metaprops))
34			return property_tuples
35
36
37			def get_metaprops(vertex_property, mapping):
38			props = mapping.ogm_properties
39			metaprops = {}
40			for ogm_name, (db_name, data_type) in props.items():
41			val = getattr(vertex_property, ogm_name, None)
42			metaprops[db_name] = data_type.to_db(val)
43			return metaprops
44
45
46			def map_vertex_to_ogm(result, props, element, *, mapping=None):
47			"""Map a vertex returned by DB to OGM vertex"""
48			props.pop('id')
49			label = props.pop('label')
50			for db_name, value in props.items():
51			metaprops = []
52			if len(value) > 1:
53			values = []
54			for v in value:
55			if isinstance(v, dict):
56			val = v.pop('value')
57			v.pop('key')
58			vid = v.pop('id')
59			if v:
60			v['id'] = vid
61			metaprops.append((val, v))
62			values.append(val)
63			else:
64			values.append(v)
65			value = values
66			else:
67			value = value[0]
68			if isinstance(value, dict):
69			val = value.pop('value')
70			value.pop('key')
71			vid = value.pop('id')
72			if value:
73			value['id'] = vid
74			metaprops.append((val, value))
75			value = val
76			name, data_type = mapping.db_properties.get(db_name, (db_name, None))
77			if data_type:
78			value = data_type.to_ogm(value)
79			setattr(element, name, value)
80			if metaprops:
81			vert_prop = getattr(element, name)
82			if hasattr(vert_prop, 'mapper_func'):
83			# Temporary hack for managers
84			vert_prop.mapper_func(metaprops, vert_prop)
85			else:
86			vert_prop.__mapping__.mapper_func(metaprops, vert_prop)
87			setattr(element, '__label__', label)
88			setattr(element, 'id', result.id)
89			return element
90
91
92			# TODO: temp hack
93			def get_hashable_id(val: Dict[str, Any]) -> Any:
94			if isinstance(val, dict) and "@type" in val and "@value" in val:
95			if val["@type"] == "janusgraph:RelationIdentifier":
96			val = val["@value"].get("value", val["@value"]["relationId"])
97			return val
98
99
100
101			def map_vertex_property_to_ogm(result, element, *, mapping=None):
102			"""Map a vertex returned by DB to OGM vertex"""
103			for (val, metaprops) in result:
104			if isinstance(element, list):
105			current = element.vp_map.get(get_hashable_id(metaprops['id']))
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-01-30 17:37 UTC by Report Bug Copy Issue Report The variable `get_hashable_id` does not seem to be defined. Loading history...
106			# This whole system needs to be reevaluated
107			if not current:
108			current = element(val)
109			if isinstance(current, list):
110			for vp in current:
111			if not hasattr(vp, '_id'):
112			element.vp_map[get_hashable_id(
113			metaprops['id'])] = vp
114			current = vp
115			break
116			elif isinstance(element, set):
117			current = element(val)
118			else:
119			current = element
120			for db_name, value in metaprops.items():
121			name, data_type = mapping.db_properties.get(
122			db_name, (db_name, None))
123			if data_type:
124			value = data_type.to_ogm(value)
125			setattr(current, name, value)
126
127
128			def map_edge_to_ogm(result, props, element, *, mapping=None):
129			"""Map an edge returned by DB to OGM edge"""
130			props.pop(T.id)
131			label = props.pop(T.label)
132			for db_name, value in props.items():
133			name, data_type = mapping.db_properties.get(db_name, (db_name, None))
134			if data_type:
135			value = data_type.to_ogm(value)
136			setattr(element, name, value)
137			setattr(element, '__label__', label)
138			setattr(element, 'id', result.id)
139			# Currently not included in graphson
140			# setattr(element.source, '__label__', result.outV.label)
141			# setattr(element.target, '__label__', result.inV.label)
142			sid = result.outV.id
143			esid = getattr(element.source, 'id', None)
144			if _check_id(sid, esid):
145			from goblin.element import GenericVertex
146			element.source = GenericVertex()
147			tid = result.inV.id
148			etid = getattr(element.target, 'id', None)
149			if _check_id(tid, etid):
150			from goblin.element import GenericVertex
151			element.target = GenericVertex()
152			setattr(element.source, 'id', sid)
153			setattr(element.target, 'id', tid)
154			return element
155
156
157			def _check_id(rid, eid):
158			if eid and rid != eid:
159			logger.warning('Edge vertex id has changed')
160			return True
161			return False
162
163
164			# DB <-> OGM Mapping
165			def create_mapping(namespace, properties):
166			"""Constructor for :py:class:`Mapping`"""
167			element_type = namespace['__type__']
168			if element_type == 'vertex':
169			mapping_func = map_vertex_to_ogm
170			mapping = Mapping(namespace, element_type, mapping_func, properties)
171			elif element_type == 'edge':
172			mapping_func = map_edge_to_ogm
173			mapping = Mapping(namespace, element_type, mapping_func, properties)
174			elif element_type == 'vertexproperty':
175			mapping_func = map_vertex_property_to_ogm
176			mapping = Mapping(namespace, element_type, mapping_func, properties)
177			else:
178			mapping = None
179			return mapping
180
181
182			class Mapping:
183			"""
184			This class stores the information necessary to map between an OGM element
185			and a DB element.
186			"""
187
188			def __init__(self, namespace, element_type, mapper_func, properties):
189			self._label = namespace['__label__']
190			self._element_type = element_type
191			self._mapper_func = functools.partial(mapper_func, mapping=self)
192			self._db_properties = {}
193			self._ogm_properties = {}
194			self._map_properties(properties)
195
196			@property
197			def label(self):
198			"""Element label"""
199			return self._label
200
201			@property
202			def mapper_func(self):
203			"""Function responsible for mapping db results to ogm"""
204			return self._mapper_func
205
206			@property
207			def db_properties(self):
208			"""A dictionary of property mappings"""
209			return self._db_properties
210
211			@property
212			def ogm_properties(self):
213			"""A dictionary of property mappings"""
214			return self._ogm_properties
215
216			def __getattr__(self, value):
217			try:
218			mapping, _ = self._ogm_properties[value]
219			return mapping
220			except KeyError:
221			raise exception.MappingError(
222			"unrecognized property {} for class: {}".format(
223			value, self._element_type))
224
225			def _map_properties(self, properties):
226			for name, prop in properties.items():
227			data_type = prop.data_type
228			if prop.db_name:
229			db_name = prop.db_name
230			else:
231			db_name = name
232			if hasattr(prop, '__mapping__'):
233			if not self._element_type == 'vertex':
234			raise exception.MappingError(
235			'Only vertices can have vertex properties')
236			self._db_properties[db_name] = (name, data_type)
237			self._ogm_properties[name] = (db_name, data_type)
238
239			def __repr__(self):
240			return '<{}(type={}, label={}, properties={})>'.format(
241			self.__class__.__name__, self._element_type, self._label,
242			self._ogm_properties)
243

goblin-ogm / goblin

goblin.mapper.map_props_to_db() B last analyzed 2022-05-13 18:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

goblin.mapper.map_props_to_db() B
last analyzed 2022-05-13 18:52 UTC