Coverage for /var/devmt/py/docp-dbi_1.1.0/docp_dbi/libs/utilities.py: 100%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3""" 

4:Purpose: This module provides generalised utility-based functionality 

5 for the project. 

6 

7:Platform: Linux/Windows | Python 3.11+ 

8:Developer: J Berendt 

9:Email: development@s3dev.uk 

10 

11:Comments: n/a 

12 

13""" 

14 

15from __future__ import annotations 

16 

17import csv 

18import os 

19# locals 

20try: 

21 from ..objects.neo4jobjects import Node, Edge 

22except ImportError: 

23 from docp_dbi.objects.neo4jobjects import Node, Edge 

24 

25 

26class Utilities: 

27 """Generalised project-based utilities.""" 

28 

29 @staticmethod 

30 def csv_to_nodes(path: str) -> list[Node]: 

31 """Create a list of :class:`Node` objects from CSV file. 

32 

33 Args: 

34 path (str): Full path to the input file. 

35 

36 .. important:: 

37 

38 The CSV file is expected to have the following field names: 

39 - id 

40 - label 

41 

42 Raises: 

43 FileNotFoundError: Raised if the provided file path does not 

44 exist. 

45 

46 Returns: 

47 list[Node]: A list of :class:`Node` objects, created from the 

48 CSV file. 

49 

50 """ 

51 if not os.path.exists(path): 

52 raise FileNotFoundError(f'Source file not found: {path}') 

53 nodes = [] 

54 with open(path, encoding='utf-8') as f: 

55 for rec in csv.DictReader(f): 

56 id_ = rec.pop('id') 

57 label = rec.pop('label') 

58 nodes.append(Node(id=id_, label=label, properties=rec)) 

59 return nodes 

60 

61 @staticmethod 

62 def csv_to_edges(path: str) -> list[Edge]: 

63 """Create a list of :class:`Edge` objects from CSV file. 

64 

65 Args: 

66 path (str): Full path to the input file. 

67 

68 .. important:: 

69 

70 The CSV file is expected to have the following field names: 

71 - source 

72 - target 

73 - type 

74 

75 Raises: 

76 FileNotFoundError: Raised if the provided file path does not 

77 exist. 

78 

79 Returns: 

80 list[Edge]: A list of :class:`Edge` objects, created from 

81 the CSV file. 

82 

83 """ 

84 # pylint: disable=redefined-builtin # Needed here. 

85 if not os.path.exists(path): 

86 raise FileNotFoundError(f'Source file not found: {path}') 

87 edges = [] 

88 with open(path, encoding='utf-8') as f: 

89 for rec in csv.DictReader(f): 

90 source = rec.pop('source') 

91 target = rec.pop('target') 

92 type_ = rec.pop('type') 

93 edges.append(Edge(source=source, 

94 target=target, 

95 type=type_, 

96 properties=rec)) 

97 return edges 

98 

99 def merge_node_query(self, node: Node) -> str: # noqa # pylint: disable=undefined-variable 

100 """Build a MERGE node query. 

101 

102 Args: 

103 node (Node): Node object for which the query is to be built. 

104 

105 Returns: 

106 str: A MERGE Cypher query for adding a new node to the graph. 

107 

108 """ 

109 if node.properties: 

110 return self._node_query_with_properties(node=node) 

111 return self._node_query(node=node) # nocover 

112 

113 def merge_edge_query(self, edge: Edge) -> str: # noqa # pylint: disable=undefined-variable 

114 """Build a MERGE edge (relationship) query. 

115 

116 Args: 

117 edge (Edge): Edge object for which the query is to be built. 

118 

119 Returns: 

120 str: A MERGE Cypher query for adding a new edge (relationship) 

121 to the graph. 

122 

123 """ 

124 if edge.properties: 

125 return self._edge_query_with_properties(edge=edge) 

126 return self._edge_query(edge=edge) 

127 

128 @staticmethod 

129 def _node_query(node: Node) -> str: # noqa # pylint: disable=undefined-variable # nocover 

130 """Build a MERGE node query.""" 

131 q = f""" 

132 MERGE (n:{node.label} { id: {node.id} } ) 

133 RETURN n 

134 """ 

135 return q.strip() 

136 

137 @staticmethod 

138 def _node_query_with_properties(node: Node) -> str: # noqa # pylint: disable=undefined-variable 

139 """Build a MERGE node query, for a node with properties.""" 

140 # pylint: disable=consider-using-f-string # Nope. Need this. 

141 properties_ = ', '.join(map('{0}: ${0}'.format, node.properties)) 

142 q = f""" 

143 MERGE (n:{node.label} { id: {node.id} } ) 

144 SET n += { {properties_} } 

145 RETURN n 

146 """ 

147 return q.strip() 

148 

149 @staticmethod 

150 def _edge_query(edge: Edge) -> str: # noqa # pylint: disable=undefined-variable 

151 """Build a MERGE edge query.""" 

152 q = f""" 

153 MATCH (s { id: {edge.source.id} } ) 

154 MATCH (t { id: {edge.target.id} } ) 

155 MERGE (s)-[e:{edge.type}]-(t) 

156 RETURN e 

157 """ 

158 return q.strip() 

159 

160 @staticmethod 

161 def _edge_query_with_properties(edge: Edge) -> str: # noqa # pylint: disable=undefined-variable 

162 """Build a MERGE edge query, for an edge with properties.""" 

163 # pylint: disable=consider-using-f-string # Nope. Need this. 

164 properties_ = ', '.join(map('{0}: ${0}'.format, edge.properties)) 

165 q = f""" 

166 MATCH (s { id: {edge.source.id} } ) 

167 MATCH (t { id: {edge.target.id} } ) 

168 MERGE (s)-[e:{edge.type}]-(t) 

169 SET e += { { properties_} } 

170 RETURN e 

171 """ 

172 return q.strip() 

173 

174 

175utilities = Utilities()