graphinate.modeling.GraphModel.node() - Code Metrics - Inspection of "⚡ Bolt: Optimize element creation loop in modeling..." - erivlis/graphinate - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#37)

unknown

created 2026-01-10 16:04 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 collections import defaultdict, namedtuple
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from enum import Enum, auto
from functools import lru_cache
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=128)
def _get_namedtuple_class(type_name: str, field_names: tuple[str] | str) -> type:
    return namedtuple(type_name, field_names)


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

    Args:
        element_type:
        field_names:

    Returns:
        Element Supplier Callable
    """
    if element_type and field_names:
        if isinstance(field_names, str):
            return _get_namedtuple_class(element_type, field_names)
        return _get_namedtuple_class(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.
    """
    is_dynamic_type = callable(element_type)
    static_create_element = None

    if not is_dynamic_type:
        _type = element_type
        # Eagerly validate static types.
        # Note: This will raise AttributeError if _type is None (consistent with previous behavior, but happens earlier)
        # and ValueError if _type is invalid identifier.
        if not _type.isidentifier():
            raise ValueError(f"Invalid Type: {_type}. Must be a valid Python identifier.")

        static_create_element = element(_type, getters.keys())

    for item in iterable:
        if is_dynamic_type:
            _type = element_type(item)
            if not _type.isidentifier():
                raise ValueError(f"Invalid Type: {_type}. Must be a valid Python identifier.")
            create_element = element(_type, getters.keys())
        else:
            create_element = static_create_element

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

erivlis / graphinate

Pull Request — main (#37)

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