graphinate.modeling.model() - Code Metrics - Inspection of "refactoring" - erivlis/graphinate - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( dd4dc2...24a3de )

by Eran

created 2023-09-29 07:19 UTC

graphinate.modeling.model() A

↳ Parent: graphinate.modeling

Complexity

Conditions

Size

Total Lines	10
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	10
rs	10
c	0
b	0
f	0

import inspect
from collections import defaultdict, namedtuple
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from typing import Any, Callable, Optional, Union

from .typing import Edge, Edges, Element, Extractor, Items, Node, Nodes, NodeTypeAbsoluteId

UNIVERSE_NODE = None


class GraphModelError(Exception):
    pass


def element(element_type: Optional[str], field_names: Optional[Iterable[str]] = None) -> Callable[[...], Element]:
    """Create a Graph Element

    Args:
        element_type:
        field_names:

    Returns:
        Element Supplier Function
    """
    return namedtuple(element_type, field_names) if element_type and field_names else tuple


def extractor(obj: Any, key: Optional[Extractor] = None) -> Optional[str]:
    """Extract data item from Element

    Args:
        obj:
        key:

    Returns:
        Element data item
    """
    if key is None:
        return obj

    if callable(key):
        return key(obj)

    if isinstance(obj, Mapping) and isinstance(key, str):
        return obj.get(key, key)

    return key


def elements(iterable: Iterable[Any],
             element_type: Optional[Extractor] = None,
             **getters: Extractor) -> Iterable[Element]:
    """Abstract Generator of Graph elements (nodes or edges)

    Args:
        iterable: source of payload
        element_type: Optional[Extractor] source of type of the element. Defaults to Element Type name.
        getters: Extractor node field sources

    Returns:
        Iterable of Elements.
    """

    if callable(element_type):
        for item in iterable:
            create_element = element(element_type(item), getters.keys())
            kwargs = {k: extractor(item, v) for k, v in getters.items()}
            yield create_element(**kwargs)
    else:
        create_element = element(element_type, getters.keys())
        for item in iterable:
            kwargs = {k: extractor(item, v) for k, v in getters.items()}
            yield create_element(**kwargs)


@dataclass
class NodeModel:
    """Represents a Node Model

    Args:
        type: the type of the Node
        parent_type: the type of the node's parent
        uniqueness: is the Node universally unique
        parameters: parameters
        generator: Nodes generator method
        label: label source

    Properties:
        absolute_id: return the NodeModel absolute_id.
    """

    type: str
    parent_type: Optional[str] = UNIVERSE_NODE
    uniqueness: bool = False
    parameters: Optional[set[str]] = None
    generator: Optional[Nodes] = None
    label: Callable[[Any], Optional[str]] = None

    @property
    def absolute_id(self) -> NodeTypeAbsoluteId:
        return self.parent_type, self.type


class GraphModel:
    """A Graph Model

       Used to declaratively register Edge and/or Node data supplier functions by using
       decorators.
    """

    def __init__(self, name: str):
        """Create a graph model
        Parameters:
            name: the archetype name for Graphs generated based on the GraphModel.
        """
        self.name: str = name
        self._node_models: dict[NodeTypeAbsoluteId, NodeModel] = {}
        self._node_children: dict[str, list[str]] = defaultdict(list)
        self._edge_generators: dict[str, list[Edges]] = defaultdict(list)
        self._networkx_graph = None

    def __add__(self, other: 'GraphModel'):
        graph_model = GraphModel(name=f"{self.name} + {other.name}")
        for model in (self, other):
            graph_model._node_models.update(model._node_models.copy())
            graph_model._node_children.update(model._node_children.copy())
            graph_model._edge_generators.update(model._edge_generators.copy())
        return graph_model

    @property
    def node_models(self) -> dict[NodeTypeAbsoluteId, NodeModel]:
        """
        Returns:
            NodeModel for Node Types. Key values are NodeTypeAbsoluteId.
        """
        return self._node_models

    @property
    def edge_generators(self):
        """
        Returns:
            Edge generator functions for Edge Types
        """
        return self._edge_generators

    @property
    def node_types(self) -> set[str]:
        """
        Returns:
            Node Types
        """
        return {v.type for v in self._node_models.values()}

    def node_children_types(self, _type: str = UNIVERSE_NODE) -> dict[str, list[str]]:
        """Children Node Types for given input Node Type

        Parameters:
            _type:  Node Type. Default value is UNIVERSE_NODE.

        Returns:
            List of children Node Types.
        """
        return {k: v for k, v in self._node_children.items() if k == _type}

    def _validate_node_parameters(self, parameters: list[str]):
        node_types = self.node_types
        if not all(p.endswith('_id') and p == p.lower() and p[:-3] in node_types for p in parameters):
            msg = ("Illegal Arguments. Argument should conform to the following rules: "
                   "1) lowercase "
                   "2) end with '_id' "
                   "3) start with value that exists as registered node type")

            raise GraphModelError(msg)

    def node(self,
             _type: Optional[Extractor] = None,
             parent_type: Optional[str] = UNIVERSE_NODE,
             uniqueness: bool = False,
             key: Optional[Extractor] = None,
             value: Optional[Extractor] = None,
             label: Optional[Extractor] = None) -> Callable[[Items], None]:
        """Decorator to Register a Generator of node payloads as a source to create Graph Nodes. It creates a
        NodeModel object.

        Parameters:
            _type: Optional source for the Node Type. Defaults to use Generator function
                   name as the Node Type.
            parent_type: Optional parent Node Type. Defaults to UNIVERSE_NODE
            uniqueness: Is the generated Node ID universally unique. Defaults to False.
            key: Optional source for Node IDs. Defaults to use the complete Node payload
                 as Node ID.
            value: Optional source for Node value field. Defaults to use the complete
                   Node payload as Node ID.
            label: Optional source for Node label field. Defaults to use a 'str'
                   representation of the complete Node payload.

        Returns:
            None
        """

        def register_node(f: Items):
            node_type = _type or f.__name__
            model_type = f.__name__ if callable(node_type) else node_type

            def node_generator(**kwargs) -> Iterable[Node]:
                yield from elements(f(**kwargs), node_type, key=key, value=value)

            parameters = inspect.getfullargspec(f).args
            node_model = NodeModel(type=model_type,
                                   parent_type=parent_type,
                                   uniqueness=uniqueness,
                                   parameters=set(parameters),
                                   label=label,
                                   generator=node_generator)
            self._node_models[node_model.absolute_id] = node_model
            self._node_children[parent_type].append(model_type)

            self._validate_node_parameters(parameters)

        return register_node

    def edge(self,
             _type: Optional[str] = None,
             source: Extractor = 'source',
             target: Extractor = 'target',
             label: Optional[Extractor] = str,
             value: Optional[Extractor] = None,
             weight: Union[float, Callable[[Any], float]] = 1.0,
             ) -> Callable[[Items], None]:
        """Decorator to Register a generator of edge payloads as a source to create
           Graph Edges. It creates an Edge generator function.

        Parameters:
            _type: Optional source for the Node Type. Defaults to use Generator function
                   name as the Node Type.
            source: Source for edge source Node ID.
            target: Source for edge target Node ID.
            label: Source for edge label.
            value: Source for edge value.
            weight: Source for edge weight.

        Returns:
            None.
        """

        def register_edge(f: Items):
            edge_type = _type or f.__name__

            getters = {
                'source': source,
                'target': target,
                'label': label,
                'type': edge_type,
                'value': value,
                'weight': weight
            }

            def edge_generator(**kwargs) -> Iterable[Edge]:
                yield from elements(f(**kwargs), edge_type, **getters)

            self._edge_generators[edge_type].append(edge_generator)

        return register_edge

    def rectify(self, _type: Optional[Extractor] = None,
                parent_type: Optional[str] = UNIVERSE_NODE,
                key: Optional[Extractor] = None,
                value: Optional[Extractor] = None,
                label: Optional[Extractor] = None):
        """Rectify the model.
           Add a default NodeModel in case of having just edge supplier/s and no node supplier/s.

           Parameters:
               _type
               parent_type
               key
               value
               label

           Returns:
               None
        """
        if self._edge_generators and not self._node_models:
            @self.node(
                _type=_type or 'node',
                parent_type=parent_type or 'node',
                uniqueness=True,
                key=key,
                value=value,
                label=label or str
            )
            def node():  # pragma: no cover
                return
                yield


def model(name: str):
    """Create a graph model

    Args:
        name: model name

    Returns:
        GraphModel
    """
    return GraphModel(name=name)


__all__ = ('GraphModel', 'UNIVERSE_NODE', 'model')


1			import inspect
2			from collections import defaultdict, namedtuple
3			from collections.abc import Iterable, Mapping
4			from dataclasses import dataclass
5			from typing import Any, Callable, Optional, Union
6
7			from .typing import Edge, Edges, Element, Extractor, Items, Node, Nodes, NodeTypeAbsoluteId
8
9			UNIVERSE_NODE = None
10
11
12			class GraphModelError(Exception):
13			pass
14
15
16			def element(element_type: Optional[str], field_names: Optional[Iterable[str]] = None) -> Callable[[...], Element]:
17			"""Create a Graph Element
18
19			Args:
20			element_type:
21			field_names:
22
23			Returns:
24			Element Supplier Function
25			"""
26			return namedtuple(element_type, field_names) if element_type and field_names else tuple
27
28
29			def extractor(obj: Any, key: Optional[Extractor] = None) -> Optional[str]:
30			"""Extract data item from Element
31
32			Args:
33			obj:
34			key:
35
36			Returns:
37			Element data item
38			"""
39			if key is None:
40			return obj
41
42			if callable(key):
43			return key(obj)
44
45			if isinstance(obj, Mapping) and isinstance(key, str):
46			return obj.get(key, key)
47
48			return key
49
50
51			def elements(iterable: Iterable[Any],
52			element_type: Optional[Extractor] = None,
53			**getters: Extractor) -> Iterable[Element]:
54			"""Abstract Generator of Graph elements (nodes or edges)
55
56			Args:
57			iterable: source of payload
58			element_type: Optional[Extractor] source of type of the element. Defaults to Element Type name.
59			getters: Extractor node field sources
60
61			Returns:
62			Iterable of Elements.
63			"""
64
65			if callable(element_type):
66			for item in iterable:
67			create_element = element(element_type(item), getters.keys())
68			kwargs = {k: extractor(item, v) for k, v in getters.items()}
69			yield create_element(**kwargs)
70			else:
71			create_element = element(element_type, getters.keys())
72			for item in iterable:
73			kwargs = {k: extractor(item, v) for k, v in getters.items()}
74			yield create_element(**kwargs)
75
76
77			@dataclass
78			class NodeModel:
79			"""Represents a Node Model
80
81			Args:
82			type: the type of the Node
83			parent_type: the type of the node's parent
84			uniqueness: is the Node universally unique
85			parameters: parameters
86			generator: Nodes generator method
87			label: label source
88
89			Properties:
90			absolute_id: return the NodeModel absolute_id.
91			"""
92
93			type: str
94			parent_type: Optional[str] = UNIVERSE_NODE
95			uniqueness: bool = False
96			parameters: Optional[set[str]] = None
97			generator: Optional[Nodes] = None
98			label: Callable[[Any], Optional[str]] = None
99
100			@property
101			def absolute_id(self) -> NodeTypeAbsoluteId:
102			return self.parent_type, self.type
103
104
105			class GraphModel:
106			"""A Graph Model
107
108			Used to declaratively register Edge and/or Node data supplier functions by using
109			decorators.
110			"""
111
112			def __init__(self, name: str):
113			"""Create a graph model
114			Parameters:
115			name: the archetype name for Graphs generated based on the GraphModel.
116			"""
117			self.name: str = name
118			self._node_models: dict[NodeTypeAbsoluteId, NodeModel] = {}
119			self._node_children: dict[str, list[str]] = defaultdict(list)
120			self._edge_generators: dict[str, list[Edges]] = defaultdict(list)
121			self._networkx_graph = None
122
123			def __add__(self, other: 'GraphModel'):
124			graph_model = GraphModel(name=f"{self.name} + {other.name}")
125			for model in (self, other):
126			graph_model._node_models.update(model._node_models.copy())
127			graph_model._node_children.update(model._node_children.copy())
128			graph_model._edge_generators.update(model._edge_generators.copy())
129			return graph_model
130
131			@property
132			def node_models(self) -> dict[NodeTypeAbsoluteId, NodeModel]:
133			"""
134			Returns:
135			NodeModel for Node Types. Key values are NodeTypeAbsoluteId.
136			"""
137			return self._node_models
138
139			@property
140			def edge_generators(self):
141			"""
142			Returns:
143			Edge generator functions for Edge Types
144			"""
145			return self._edge_generators
146
147			@property
148			def node_types(self) -> set[str]:
149			"""
150			Returns:
151			Node Types
152			"""
153			return {v.type for v in self._node_models.values()}
154
155			def node_children_types(self, _type: str = UNIVERSE_NODE) -> dict[str, list[str]]:
156			"""Children Node Types for given input Node Type
157
158			Parameters:
159			_type: Node Type. Default value is UNIVERSE_NODE.
160
161			Returns:
162			List of children Node Types.
163			"""
164			return {k: v for k, v in self._node_children.items() if k == _type}
165
166			def _validate_node_parameters(self, parameters: list[str]):
167			node_types = self.node_types
168			if not all(p.endswith('_id') and p == p.lower() and p[:-3] in node_types for p in parameters):
169			msg = ("Illegal Arguments. Argument should conform to the following rules: "
170			"1) lowercase "
171			"2) end with '_id' "
172			"3) start with value that exists as registered node type")
173
174			raise GraphModelError(msg)
175
176			def node(self,
177			_type: Optional[Extractor] = None,
178			parent_type: Optional[str] = UNIVERSE_NODE,
179			uniqueness: bool = False,
180			key: Optional[Extractor] = None,
181			value: Optional[Extractor] = None,
182			label: Optional[Extractor] = None) -> Callable[[Items], None]:
183			"""Decorator to Register a Generator of node payloads as a source to create Graph Nodes. It creates a
184			NodeModel object.
185
186			Parameters:
187			_type: Optional source for the Node Type. Defaults to use Generator function
188			name as the Node Type.
189			parent_type: Optional parent Node Type. Defaults to UNIVERSE_NODE
190			uniqueness: Is the generated Node ID universally unique. Defaults to False.
191			key: Optional source for Node IDs. Defaults to use the complete Node payload
192			as Node ID.
193			value: Optional source for Node value field. Defaults to use the complete
194			Node payload as Node ID.
195			label: Optional source for Node label field. Defaults to use a 'str'
196			representation of the complete Node payload.
197
198			Returns:
199			None
200			"""
201
202			def register_node(f: Items):
203			node_type = _type or f.__name__
204			model_type = f.__name__ if callable(node_type) else node_type
205
206			def node_generator(**kwargs) -> Iterable[Node]:
207			yield from elements(f(**kwargs), node_type, key=key, value=value)
208
209			parameters = inspect.getfullargspec(f).args
210			node_model = NodeModel(type=model_type,
211			parent_type=parent_type,
212			uniqueness=uniqueness,
213			parameters=set(parameters),
214			label=label,
215			generator=node_generator)
216			self._node_models[node_model.absolute_id] = node_model
217			self._node_children[parent_type].append(model_type)
218
219			self._validate_node_parameters(parameters)
220
221			return register_node
222
223			def edge(self,
224			_type: Optional[str] = None,
225			source: Extractor = 'source',
226			target: Extractor = 'target',
227			label: Optional[Extractor] = str,
228			value: Optional[Extractor] = None,
229			weight: Union[float, Callable[[Any], float]] = 1.0,
230			) -> Callable[[Items], None]:
231			"""Decorator to Register a generator of edge payloads as a source to create
232			Graph Edges. It creates an Edge generator function.
233
234			Parameters:
235			_type: Optional source for the Node Type. Defaults to use Generator function
236			name as the Node Type.
237			source: Source for edge source Node ID.
238			target: Source for edge target Node ID.
239			label: Source for edge label.
240			value: Source for edge value.
241			weight: Source for edge weight.
242
243			Returns:
244			None.
245			"""
246
247			def register_edge(f: Items):
248			edge_type = _type or f.__name__
249
250			getters = {
251			'source': source,
252			'target': target,
253			'label': label,
254			'type': edge_type,
255			'value': value,
256			'weight': weight
257			}
258
259			def edge_generator(**kwargs) -> Iterable[Edge]:
260			yield from elements(f(kwargs), edge_type, getters)
261
262			self._edge_generators[edge_type].append(edge_generator)
263
264			return register_edge
265
266			def rectify(self, _type: Optional[Extractor] = None,
267			parent_type: Optional[str] = UNIVERSE_NODE,
268			key: Optional[Extractor] = None,
269			value: Optional[Extractor] = None,
270			label: Optional[Extractor] = None):
271			"""Rectify the model.
272			Add a default NodeModel in case of having just edge supplier/s and no node supplier/s.
273
274			Parameters:
275			_type
276			parent_type
277			key
278			value
279			label
280
281			Returns:
282			None
283			"""
284			if self._edge_generators and not self._node_models:
285			@self.node(
286			_type=_type or 'node',
287			parent_type=parent_type or 'node',
288			uniqueness=True,
289			key=key,
290			value=value,
291			label=label or str
292			)
293			def node(): # pragma: no cover
294			return
295			yield
296
297
298			def model(name: str):
299			"""Create a graph model
300
301			Args:
302			name: model name
303
304			Returns:
305			GraphModel
306			"""
307			return GraphModel(name=name)
308
309
310			__all__ = ('GraphModel', 'UNIVERSE_NODE', 'model')
311

erivlis / graphinate

Push — main ( dd4dc2...24a3de )

graphinate.modeling.model() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like