Coverage for /var/devmt/py/docp-dbi_1.1.0/docp_dbi/databases/neo4j.py: 100%

1#!/usr/bin/env python3 

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

3""" 

4:Purpose: This module provides a :class:`Graph` class for interacting 

5 with a Neo4j graph database. 

6 

7:Platform: Linux/Windows | Python 3.11+ 

8:Developer: J Berendt 

9:Email: development@s3dev.uk 

10 

11:Comments: n/a 

12 

13:Example: 

14 

15 Example showing how to generate a ``.env`` file for easy connection:: 

16 

17 >>> from docp_dbi.databases import neo4j 

18 

19 >>> neo4j.create_dotenv_file(uri='neo4j://127.0.0.1:7687', 

20 database='neo4j', 

21 username='neo4j', 

22 password='neo4j') 

23 

24 # Copy the .env file from your Desktop to the project's 

25 # directory. 

26 

27 

28 Example showing how to use the *local* ``.env`` file to connect to 

29 the database:: 

30 

31 >>> from docp_dbi import Neo4jDB 

32 >>> neodb = Neo4jDB() 

33 

34 # Verify the database is connected. 

35 >>> neodb.connected 

36 True 

37 

38 # Query the database. 

39 >>> results = neodb.execute_query('match (n) return count(n)') 

40 

41 

42 Example showing how to use the ``.env`` *filepath* to connect to 

43 the database:: 

44 

45 >>> from docp_dbi import Neo4jDB 

46 >>> neodb = Neo4jDB(auth='/path/to/.env') 

47 

48 # Verify the database is connected. 

49 >>> neodb.connected 

50 True 

51 

52 # Query the database. 

53 >>> results = neodb.execute_query('match (n) return count(n)') 

54 

55 

56 Example showing how to query a graph database, using the URI and 

57 credentials:: 

58 

59 >>> from docp_dbi import Neo4jDB 

60 >>> neodb = Neo4jDB(uri='neo4j://127.0.0.1:7687', 

61 auth=('username', 'password')) 

62 

63 # Verify the database is connected. 

64 >>> neodb.connected 

65 True 

66 

67 # Query the database. 

68 >>> results = neodb.execute_query('match (n) return count(n)') 

69 

70""" 

71# pylint: disable=import-error # Spurious 

72# pylint: disable=no-name-in-module # Spurious: neo4j._sync 

73# pylint: disable=wrong-import-order 

74 

75from __future__ import annotations 

76 

77import dotenv 

78import os 

79# Imported this way as this module is also called neo4j. 

80from neo4j._sync.driver import GraphDatabase as _GraphDatabase 

81# locals 

82try: 

83 from ..libs.utilities import utilities 

84 from ..objects.neo4jobjects import GraphObject 

85except ImportError: 

86 from docp_dbi.libs.utilities import utilities 

87 from docp_dbi.objects.neo4jobjects import GraphObject 

88 

89 

90class Neo4jDB: 

91 """Graph object class for interacting with a Neo4j graph database. 

92 

93 Args: 

94 uri (str, optional): Database connection URI. Defaults to None. 

95 auth (str | tuple, optional): Either a path to the ``.env`` file 

96 containing database credentials, or a tuple containing 

97 authentication credentials as ``(username, password)``. 

98 If None, this parameter's value is altered to search the 

99 current directory for a ``.env`` file. Defaults to None. 

100 database (str, optional): Name of the database to be used. Can 

101 also be provided via the ``.env`` file using the 

102 key: ``NEO4J_DATABASE``. Defaults to None. 

103 **config: Pass-through configuration items provided to the driver. 

104 

105 .. note:: 

106 If values are not provided to the class constructor, an attempt 

107 is made to obtain the values from the ``.env`` file, either in 

108 the current directory, or from the path provided to the ``auth`` 

109 parameter. 

110 

111 """ 

112 

113 _DIR_ROOT = os.path.dirname(os.path.realpath(__file__)) 

114 

115 def __init__(self, 

116 uri: str=None, 

117 *, 

118 auth: str | tuple=None, 

119 database: str=None, 

120 **config) -> None: 

121 """Graph class initialiser.""" 

122 self._uri = uri 

123 self._auth = auth 

124 self._db = database 

125 self._config = config 

126 self._connected = False 

127 self._driver = None 

128 self._connect() 

129 

130 def __del__(self) -> None: 

131 """Class deconstructor.""" 

132 # pylint: disable=broad-exception-caught # OK, it's a pass. 

133 try: 

134 self._driver.close() 

135 self._connected = False 

136 except Exception: # nocover 

137 pass 

138 

139 @property 

140 def connected(self) -> bool: 

141 """The database is connected. 

142 

143 Returns: 

144 bool: True if connected, otherwise False. 

145 

146 """ 

147 return self._connected 

148 

149 @property 

150 def database(self) -> str: 

151 """Accessor to the database name. 

152 

153 The database name can be provided using the ``.env`` file with 

154 the key: ``NEO4J_DATABASE``. 

155 

156 """ 

157 return self._db 

158 

159 @property 

160 def driver(self) -> Neo4jDriver: # noqa # pylint: disable=undefined-variable 

161 """Accessor to the database driver object.""" 

162 return self._driver 

163 

164 def close(self) -> None: 

165 """Close the connection.""" 

166 self._driver.close() 

167 self._connected = False 

168 

169 def execute_query(self, 

170 query: str, 

171 parameters: dict=None, 

172 database: str=None, 

173 **kwargs) -> EagerResult: # noqa # pylint: disable=undefined-variable 

174 """Execute a query on the database. 

175 

176 Args: 

177 query (str): Cypher query to be executed. 

178 parameters (dict, optional): Query parameters. 

179 Defaults to None. 

180 database (str, optional): Database to be used. 

181 This parameter can be used to override the class' 

182 database attribute (:attr:`database`), if provided via 

183 the ``.env`` file. Defaults to None. 

184 **kwargs: Pass-through keyword arguments. 

185 

186 Returns: 

187 EagerResult: Query results object. 

188 

189 """ 

190 return self._driver.execute_query(query_=query, 

191 parameters_=parameters, 

192 database_=database or self._db, 

193 **kwargs) 

194 

195 def load_from_csv(self, 

196 *, 

197 path_nodes: str=None, 

198 path_edges: str=None) -> tuple[bool, list]: 

199 """Load graph data from a CSV file. 

200 

201 This function is designed to load node and/or edge (relationship) 

202 data. The required format for each CSV type is as follows, where 

203 the property data is *optional*, whereas the ``id`` and ``label`` 

204 are *required*. 

205 

206 **Node Data** 

207 

208 - **Field Definitions** 

209 - **id:** Unique (user-defined) node ID. 

210 - **label:** Node label. For example: Book, Character, etc. 

211 - **Properties:** The column heading for each property will 

212 be used as its key in the graph. 

213 

214 - **Format:** 

215 

216 ```csv 

217 id,label,prop1,...,propN 

218 ``` 

219 

220 **Edge Data** 

221 

222 - **Field Definitions** 

223 - **source:** The ID of the *source* node. 

224 - **target:** The ID of the *target* node. 

225 - **type:** edge type. For example: AUTHOR, 

226 APPEARS_IN, etc. 

227 - **Properties:** The column heading for each property will 

228 be used as its key in the graph. 

229 

230 - **Format:** 

231 

232 ```csv 

233 source,target,type,prop1,...,propN 

234 ``` 

235 

236 Args: 

237 path_nodes (str, optional): Full path to the node input data. 

238 Defaults to None. 

239 path_edges (str, optional): Full path to the edge input data. 

240 Defaults to None. 

241 

242 Returns: 

243 tuple[bool, list]: A tuple containing a success flag and a 

244 list of query result objects. 

245 

246 """ 

247 n = [] 

248 e = [] 

249 src = None 

250 if path_nodes: 

251 n = utilities.csv_to_nodes(path=path_nodes) 

252 src = os.path.basename(path_nodes) 

253 if path_edges: 

254 e = utilities.csv_to_edges(path=path_edges) 

255 if src is None: 

256 src = os.path.basename(path_edges) 

257 go = GraphObject(nodes=n, edges=e, source=src) 

258 return self._add_graph_object(graphobject=go) 

259 

260 def _add_graph_object(self, graphobject: GraphObject) -> tuple[bool, list]: # noqa 

261 """Add nodes and relationships to the graph via GraphObject. 

262 

263 Args: 

264 graphobjects (GraphObject): A :class:`GraphObject` 

265 container class wrapping the :class:`Node` and 

266 :class:`Edge` objects to be added to the graph. 

267 

268 Returns: 

269 tuple[bool, list]: A tuple containing a success flag and a 

270 list of Neo4j result objects. 

271 

272 """ 

273 results = [] 

274 for node in graphobject.nodes: 

275 q = utilities.merge_node_query(node=node) 

276 results.append(self.execute_query(q, parameters=node.properties)) 

277 for edge in graphobject.edges: 

278 q = utilities.merge_edge_query(edge=edge) 

279 results.append(self.execute_query(q, parameters=edge.properties)) 

280 act = len(results) 

281 exp = graphobject.count 

282 if exp == act: 

283 print(f'{act} nodes/edges added successfully.') 

284 else: # nocover 

285 print('[ERROR]: {act} of {exp} nodes/edges added.') 

286 return (exp == act), results 

287 

288 def _attempt_credentials(self) -> None: 

289 """Attempt to use the ``.env`` file for driver connection.""" 

290 if not all((os.path.isfile(self._auth), os.path.exists(self._auth))): 

291 raise RuntimeError('The authentication file either is not a file or ' 

292 f'cannot be found: {self._auth}') 

293 dotenv.load_dotenv(dotenv_path=self._auth) 

294 if self._uri is None: 

295 self._uri = os.environ.get('NEO4J_URI') 

296 self._auth = (os.environ.get('NEO4J_USERNAME'), os.environ.get('NEO4J_PASSWORD')) 

297 # Class constructor overrides the database name. 

298 if self._db is None: 

299 self._db = os.environ.get('NEO4J_DATABASE') 

300 

301 def _connect(self) -> None: 

302 """Connect to the database. 

303 

304 If the ``auth`` argument is a file path, an attempt is made to 

305 use the credentials from the ``.env`` file. Otherwise, the 

306 ``auth`` argument credentials tuple is used. 

307 

308 """ 

309 if self._auth is None: # nocover 

310 self._auth = os.path.join(self._DIR_ROOT, '.env') 

311 if isinstance(self._auth, str): 

312 self._attempt_credentials() 

313 self._driver = _GraphDatabase.driver(uri=self._uri, auth=self._auth, **self._config) 

314 if self._driver is None or not self._driver.verify_authentication(): 

315 raise RuntimeError('Database authentication failed. Review the credentials.') 

316 assert self._driver.verify_connectivity() is None 

317 self._connected = True 

318 

319 

320def create_dotenv_file(uri: str='neo4j://127.0.0.1:7687', 

321 database: str='neo4j', 

322 password: str='neo4j', 

323 username: str='neo4j', 

324 replace: bool=False) -> None: # nocover 

325 """Generate a ``.env`` file from the provided credentials. 

326 

327 The resulting file will be placed on the user's Desktop. 

328 

329 Args: 

330 uri (str, optional): Database location URI. 

331 Defaults to 'neo4j://127.0.0.1:7687'. 

332 database (str, optional): Database name. Defaults to 'neo4j'. 

333 password (str, optional): Authentication password. 

334 Defaults to 'neo4j'. 

335 username (str, optional): Authentication username. 

336 Defaults to 'neo4j'. 

337 replace (bool, optional): Force replace an existing ``.env`` 

338 file. Defaults to False. 

339 

340 Raises: 

341 RuntimeError: Raised if a ``.env`` file already exists on the 

342 user's Desktop *and* ``replace=False``. 

343 

344 """ 

345 path = os.path.expanduser('~/Desktop/.env') 

346 if os.path.exists(path) and not replace: 

347 raise RuntimeError('An .env file already exists on the Desktop. ' 

348 'To overwrite, pass replace=True.') 

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

350 f.write(f'NEO4J_URI="{uri}"\n') 

351 f.write(f'NEO4J_DATABASE="{database}"\n') 

352 f.write(f'NEO4J_PASSWORD="{password}"\n') 

353 f.write(f'NEO4J_USERNAME="{username}"')