graphinate.modeling.GraphModel.node() - Code Metrics - Inspection of "⚡ Bolt: Cache namedtuple creation to speed up grap..." - erivlis/graphinate - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#32)

unknown

created 2026-01-05 15:39 UTC

graphinate.modeling.GraphModel.node() A

↳ Parent: graphinate.modeling

Complexity

Conditions

Size

Total Lines	67
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	26
dl	0
loc	67
rs	9.256
c	0
b	0
f	0
cc	2
nop	8

How to fix Long Method Many Parameters

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

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


class GraphModelError(Exception):
    pass


@lru_cache(maxsize=None)
def _get_namedtuple(element_type: str, field_names: tuple[str]):
    return namedtuple(element_type, field_names)


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
    """
    if element_type and field_names:
        return _get_namedtuple(element_type, tuple(field_names))
    return 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 | UniverseNode | 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[NodeModel]] = defaultdict(list)
        self._edge_generators: dict[str, list[Callable[[], Iterable[Edge]]]] = defaultdict(list)
        self._networkx_graph = None

    def __add__(self, other: 'GraphModel') -> '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) -> Mapping[NodeTypeAbsoluteId, list[NodeModel]]:
        """
        Returns:
            NodeModel for Node Types. Key values are NodeTypeAbsoluteId.
        """
        return MappingProxyType(self._node_models)

    @property
    def edge_generators(self) -> Mapping[str, list[Callable[[], Iterable[Edge]]]]:
        """
        Returns:
            Edge generator functions for Edge Types
        """
        return MappingProxyType(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) -> Mapping[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 MappingProxyType({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.

        Generator Function Signature:
            The decorated generator function may accept arguments to receive context from parent nodes.
            These arguments MUST conform to the following strict naming convention:
            1. The argument name must be lowercase.
            2. The argument name must end with '_id'.
            3. The prefix (before '_id') must match an existing registered Node Type.

            Example:
                If you have a parent node type 'user', your child node generator can accept 'user_id'.

                @model.node(parent_type='user')
                def get_posts(user_id): ...

            Note: Arbitrary arguments (e.g., configuration flags) are currently NOT supported.

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

erivlis / graphinate

Pull Request — main (#32)

graphinate.modeling.GraphModel.node() A

Complexity

Size

Duplication

Importance

How to fix Long Method Many Parameters

Long Method

Many Parameters

Duplication Side-by-Side

Filter issues like