Coverage for /var/devmt/py/docp-dbi_1.1.0/docp_dbi/objects/neo4jobjects.py: 100%
92 statements
« prev ^ index » next coverage.py v7.8.0, created at 2026-06-16 17:52 +0100
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3"""
4:Purpose: This module provides the implementation for the Neo4j objects
5 used for loading data into a graph database.
7 .. note::
8 Generally, this module is for internal use only.
10:Platform: Linux/Windows | Python 3.10+
11:Developer: J Berendt
12:Email: development@s3dev.uk
14:Comments: n/a
16"""
18from __future__ import annotations
21class _GraphBaseUtilities:
22 """Base utilities for graph classes."""
23 # pylint: disable=no-member # Provided by child class.
25 _QUOTES = ('\'', '"')
27 def _clean_properties(self) -> None:
28 """Clean property values.
30 :Tasks:
31 - Convert integer strings to type ``int``.
33 """
34 for k, v in self._properties.items():
35 if v:
36 if all((isinstance(v, str), v.isdigit())):
37 self._properties[k] = int(v)
39 # Cannot be in libs.utilities due to circular import.
40 def _isquoted(self, s: str) -> bool:
41 """Test if a string is already quoted.
43 Args:
44 s (str): The string to be tested.
46 Returns:
47 bool: True if the string contains starting and ending quotes.
48 Otherwise, False.
50 """
51 return s[0] in self._QUOTES and s[-1] in self._QUOTES
54class Edge(_GraphBaseUtilities):
55 """Object representing an edge (a relationship) in the graph.
57 Args:
58 source (Node): The edge's source :class:`Node` object.
59 target (Node): The edge's target :class:`Node` object.
60 type (str): Edge type.
61 For example: 'PARENT', 'AUTHOR_OF', 'MEMBER_OF', etc.
63 .. note::
64 To follow Cypher convention, **upper case** is
65 recommended. Therefore, all types names are converted to
66 upper case.
68 properties (dict, optional): Any properties which are to be
69 assigned to the edge. Defaults to None.
71 """
72 # pylint: disable=redefined-builtin # type
74 def __init__(self, source: Node, target: Node, type: str, properties: dict=None) -> None:
75 """Relationship class initialiser."""
76 self._source = source
77 self._target = target
78 self._type = type.upper()
79 self._properties = properties or {}
80 self._validate()
81 self._clean_properties()
83 def __repr__(self) -> str: # nocover
84 """Object's string representation."""
85 r = (f'<source={self._source}; '
86 f'target={self._target}; '
87 f'type={self._type}; '
88 f'properties={self._properties}>')
89 return r
91 @property
92 def properties(self) -> dict:
93 """Accessor to the edge properties."""
94 return self._properties
96 @property
97 def source(self) -> Node:
98 """Accessor to the source :class:`Node` object."""
99 return self._source
101 @property
102 def target(self) -> Node:
103 """Accessor to the target :class:`Node` object."""
104 return self._target
106 @property
107 def type(self) -> str:
108 """Target to the type name."""
109 return self._type
111 def _validate(self) -> None:
112 """Validate the nodes to ensure a Node object exists.
114 If the source and target values come from a CSV file, they will
115 be of string or integer type and must be converted to a
116 :class:`Node` object before being added to the graph.
118 """
119 def _convert(val: int | str) -> int | str:
120 """Set the ID value as an integer, if possible."""
121 # pylint: disable=possibly-used-before-assignment # ValueError raised to catch this.
122 if not isinstance(val, int | str):
123 raise ValueError('Unexpected type when converting source (or) target to Node.')
124 if isinstance(val, str):
125 id_ = val if not val.isdigit() else int(val)
126 elif isinstance(val, int):
127 id_ = val
128 return Node(id=id_)
129 if not isinstance(self._source, Node):
130 self._source = _convert(val=self._source)
131 if not isinstance(self._target, Node):
132 self._target = _convert(val=self._target)
135class Node(_GraphBaseUtilities):
136 """Object representing a node in the graph.
138 Args:
139 id (int | str): User-defined ID to be assigned to the node.
140 This ID will be used for relationship linking and MATCH
141 referencing.
142 label (str, optional): Node label.
143 For example: 'Person' or 'Book'. Defaults to 'Node'.
145 .. note::
146 To follow Cypher convention, **Pascal case** is
147 recommended.
149 properties (dict, optional): Any properties which are to be
150 assigned to the node. Defaults to None.
152 """
153 # pylint: disable=redefined-builtin # id
155 def __init__(self, id: int | str, label: str='Node', properties: dict=None) -> None:
156 """Node class initialiser."""
157 self._id = self._create_id(id_=id)
158 self._label = label
159 self._properties = properties or {}
160 self._clean_properties()
162 def __repr__(self) -> str: # nocover
163 """Object's string representation."""
164 r = f'<id={self._id}; label={self._label}; properties={self._properties}>'
165 return r
167 @property
168 def id(self) -> int | str:
169 """Accessor to the node's ID."""
170 return self._id
172 @property
173 def label(self) -> str:
174 """Accessor to the node label."""
175 return self._label
177 @property
178 def properties(self) -> dict:
179 """Accessor to the node's properties as a dictionary."""
180 return self._properties
182 def _create_id(self, id_: int | str) -> int | str:
183 """Setter for the node ID.
185 Args:
186 id_ (int | str): ID to be set.
188 Returns:
189 str | int: If the ID is a numeric value and was defined as a
190 string, the value is returned as an integer type. Otherwise,
191 the a *quoted* string is returned (if not already quoted).
193 """
194 if isinstance(id_, int):
195 return id_
196 if isinstance(id_, str) and id_.isdigit():
197 return int(id_)
198 if self._isquoted(s=id_):
199 return id_
200 return f'"{id_}"'
203class GraphObject:
204 """Container for nodes and relationships for adding to the database.
206 Args:
207 nodes (list, optional): A list of :class:`Node` objects.
208 Defaults to None.
209 edges (list, optional): A list of :class:`Edge` objects.
210 Defaults to None.
211 source (str, optional): Data source. Defaults to None.
213 """
215 def __init__(self, nodes: list=None, edges: list=None, source: str=None) -> None:
216 """GraphObjects class initialiser."""
217 self._nodes = []
218 self._edges = []
219 self._source = source
220 self._set(n=nodes, e=edges)
221 self._validate()
223 @property
224 def count(self) -> int:
225 """Accessor to the count of node *and* edge objects."""
226 return sum((len(self._nodes), len(self._edges)))
228 @property
229 def nodes(self) -> list[Node]: #noqa
230 """Accessor to the :class:`Node` objects list."""
231 return self._nodes
233 @property
234 def edges(self) -> list[Edge]: # noqa
235 """Accessor to the :class:`Edge` objects list."""
236 return self._edges
238 def _validate(self) -> None:
239 """Validate the passed object types.
241 Raises:
242 TypeError: Raised if the wrong data type is detected.
244 """
245 if not all(map(lambda x: isinstance(x, Node | None), self._nodes)):
246 raise TypeError('All node-based nodes must be of type: Node. '
247 f'Got: {list(map(type, self._nodes))}')
248 if not all(map(lambda x: isinstance(x, Edge | None), self._edges)):
249 raise TypeError('All edge-based nodes must be of type: Edge. '
250 f'Got: {list(map(type, self._edges))}')
252 def _set(self, n: list[Node], e: list[Edge]) -> None: # noqa
253 """Set the passed objects as class instance attributes.
255 Args:
256 n (list[Node]): List of :class:`Node` objects.
257 r (list[Edge]): List of :class:`Edge` objects.
259 .. note::
260 If only a single class is passed into the constructor, this
261 is converted to a list before setting.
263 """
264 if n:
265 self._nodes = n if isinstance(n, list) else [n]
266 if e:
267 self._edges = e if isinstance(e, list) else [e]