graphinate.modeling.extractor() - Code Metrics - erivlis/graphinate - Measure and Improve Code Quality continuously with Scrutinizer

graphinate.modeling.extractor() A
last analyzed 2025-10-26 22:50 UTC

↳ Parent: graphinate.modeling

Complexity

Conditions

Size

Total Lines	20
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	8
nop	2
dl	0
loc	20
rs	9.3333
c	0
b	0
f	0

import inspect
import itertools
from collections import defaultdict, namedtuple
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from enum import Enum, auto
from typing import Any, Union

from .typing import Edge, Element, Extractor, Items, Node, NodeTypeAbsoluteId, UniverseNode


class GraphModelError(Exception):
    pass


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

    Args:
        element_type:
        field_names:

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


def extractor(obj: Any, key: Extractor | None = None) -> str | None:
    """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: Extractor | None = 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.
    """
    for item in iterable:
        _type = element_type(item) if element_type and callable(element_type) else element_type
        if not _type.isidentifier():
            raise ValueError(f"Invalid Type: {_type}. Must be a valid Python identifier.")

        create_element = element(_type, getters.keys())
        kwargs = {k: extractor(item, v) for k, v in getters.items()}
        yield create_element(**kwargs)


class Multiplicity(Enum):
    ADD = auto()
    ALL = auto()
    FIRST = auto()
    LAST = auto()


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

    Args:
        type: the type of the Node.
        parent_type: the type of the node's parent. Defaults to UniverseNode.
        parameters: parameters of the Node. Defaults to None.
        label: label source. Defaults to None.
        uniqueness: is the Node universally unique. Defaults to True.
        multiplicity: Multiplicity of the Node. Defaults to ALL.
        generator: Nodes generator method. Defaults to None.

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

    type: str
    parent_type: str | None = UniverseNode
    parameters: set[str] | None = None
    label: Callable[[Any], str | None] = None
    uniqueness: bool = True
    multiplicity: Multiplicity = Multiplicity.ALL
    generator: Callable[[], Iterable[Node]] | None = 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.

    Args:
        name: the archetype name for Graphs generated based on the GraphModel.
    """

    def __init__(self, name: str):
        self.name: str = name
        self._node_models: dict[NodeTypeAbsoluteId, list[NodeModel]] = defaultdict(list)
        self._node_children: dict[str, list[str]] = defaultdict(list)
        self._edge_generators: dict[str, list[Callable[[], Iterable[Edge]]]] = defaultdict(list)
        self._networkx_graph = None

    def __add__(self, other: 'GraphModel'):
        graph_model = GraphModel(name=f"{self.name} + {other.name}")
        for m in (self, other):
            for k, v in m._node_models.items():
                graph_model._node_models[k].extend(v)

            for k, v in m._node_children.items():
                graph_model._node_children[k].extend(v)

            for k, v in m._edge_generators.items():
                graph_model._edge_generators[k].extend(v)

        return graph_model

    @property
    def node_models(self) -> dict[NodeTypeAbsoluteId, list[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 itertools.chain.from_iterable(self._node_models.values())}

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

        Args:
            _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}

    @staticmethod
    def _validate_type(node_type: str):
        if not callable(node_type) and not node_type.isidentifier():
            raise ValueError(f"Invalid Type: {node_type}. Must be a valid Python identifier.")

    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_: Extractor | None = None,
             parent_type: str | None = UniverseNode,
             key: Extractor | None = None,
             value: Extractor | None = None,
             label: Extractor | None = None,
             unique: bool = True,
             multiplicity: Multiplicity = Multiplicity.ALL) -> Callable[[Items], None]:
        """Decorator to Register a Generator of node payloads as a source for Graph Nodes.
        It creates a NodeModel object.

        Args:
            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

            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.
            unique: is the Node universally unique. Defaults to True.
            multiplicity: Multiplicity of the Node. Defaults to ALL.

        Returns:
            None
        """

        def register_node(f: Items):
            node_type = type_ or f.__name__
            self._validate_type(node_type)

            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,
                                   parameters=set(parameters),
                                   label=label,
                                   uniqueness=unique,
                                   multiplicity=multiplicity,
                                   generator=node_generator)
            self._node_models[node_model.absolute_id].append(node_model)
            self._node_children[parent_type].append(model_type)

            self._validate_node_parameters(parameters)

        return register_node

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

        Args:
            type_: Optional source for the Edge Type. Defaults to use Generator function
                   name as the Edge 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__
            self._validate_type(edge_type)

            model_type = f.__name__ if callable(edge_type) else edge_type

            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[model_type].append(edge_generator)

        return register_edge

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

       Args:
           _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',
                unique=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', 'Multiplicity', 'model')


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

erivlis / graphinate

graphinate.modeling.extractor() A last analyzed 2025-10-26 22:50 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

graphinate.modeling.extractor() A
last analyzed 2025-10-26 22:50 UTC