graphinate.modeling.GraphModel.node() - Code Metrics - Inspection of "perf: Cache namedtuple class creation in modeling...." - erivlis/graphinate - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( c36ff9...86504d )

by Eran

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

erivlis / graphinate

Push — main ( c36ff9...86504d )

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