#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
:Purpose: This module provides a :class:`Graph` class for interacting
with a Neo4j graph database.
:Platform: Linux/Windows | Python 3.11+
:Developer: J Berendt
:Email: development@s3dev.uk
:Comments: n/a
:Example:
Example showing how to **generate** a ``.env`` file for easy
connection::
>>> from docp_dbi.databases import neo4j
>>> neo4j.create_dotenv_file(uri='neo4j://127.0.0.1:7687',
database='neo4j',
username='neo4j',
password='neo4j')
# Copy the .env file from your Desktop to the project's
# directory.
Example showing how to use the *local* ``.env`` file to connect to
the database::
>>> from docp_dbi import Neo4jDB
>>> neodb = Neo4jDB()
# Verify the database is connected.
>>> neodb.connected
True
# Query the database.
>>> results = neodb.execute_query('match (n) return count(n)')
Example showing how to use the ``.env`` *filepath* to connect to
the database::
>>> from docp_dbi import Neo4jDB
>>> neodb = Neo4jDB(auth='/path/to/.env')
# Verify the database is connected.
>>> neodb.connected
True
# Query the database.
>>> results = neodb.execute_query('match (n) return count(n)')
Example showing how to query a graph database, using the URI and
credentials::
>>> from docp_dbi import Neo4jDB
>>> neodb = Neo4jDB(uri='neo4j://127.0.0.1:7687',
auth=('username', 'password'))
# Verify the database is connected.
>>> neodb.connected
True
# Query the database.
>>> results = neodb.execute_query('match (n) return count(n)')
"""
# pylint: disable=import-error # Spurious
# pylint: disable=no-name-in-module # Spurious: neo4j._sync
# pylint: disable=wrong-import-order
from __future__ import annotations
import dotenv
import os
# Imported this way as this module is also called neo4j.
from neo4j._sync.driver import GraphDatabase as _GraphDatabase
# locals
try:
from ..libs.utilities import utilities
from ..objects.neo4jobjects import GraphObject
except ImportError:
from docp_dbi.libs.utilities import utilities
from docp_dbi.objects.neo4jobjects import GraphObject
[docs]
class Neo4jDB:
"""Graph object class for interacting with a Neo4j graph database.
Args:
uri (str, optional): Database connection URI. Defaults to None.
auth (str | tuple, optional): Either a path to the ``.env`` file
containing database credentials, or a tuple containing
authentication credentials as ``(username, password)``.
If None, this parameter's value is altered to search the
current directory for a ``.env`` file. Defaults to None.
database (str, optional): Name of the database to be used. Can
also be provided via the ``.env`` file using the
key: ``NEO4J_DATABASE``. Defaults to None.
**config: Pass-through configuration items provided to the driver.
.. note::
If values are not provided to the class constructor, an attempt
is made to obtain the values from the ``.env`` file, either in
the current directory, or from the path provided to the ``auth``
parameter.
"""
_DIR_ROOT = os.path.dirname(os.path.realpath(__file__))
def __init__(self,
uri: str=None,
*,
auth: str | tuple=None,
database: str=None,
**config) -> None:
"""Graph class initialiser."""
self._uri = uri
self._auth = auth
self._db = database
self._config = config
self._connected = False
self._driver = None
self._connect()
def __del__(self) -> None:
"""Class deconstructor."""
# pylint: disable=broad-exception-caught # OK, it's a pass.
try:
self._driver.close()
self._connected = False
except Exception: # nocover
pass
@property
def connected(self) -> bool:
"""The database is connected.
Returns:
bool: True if connected, otherwise False.
"""
return self._connected
@property
def database(self) -> str:
"""Accessor to the database name.
The database name can be provided using the ``.env`` file with
the key: ``NEO4J_DATABASE``.
"""
return self._db
@property
def driver(self) -> Neo4jDriver: # noqa # pylint: disable=undefined-variable
"""Accessor to the database driver object."""
return self._driver
[docs]
def close(self) -> None:
"""Close the connection."""
self._driver.close()
self._connected = False
[docs]
def execute_query(self,
query: str,
parameters: dict=None,
database: str=None,
**kwargs) -> EagerResult: # noqa # pylint: disable=undefined-variable
"""Execute a query on the database.
Args:
query (str): Cypher query to be executed.
parameters (dict, optional): Query parameters.
Defaults to None.
database (str, optional): Database to be used.
This parameter can be used to override the class'
database attribute (:attr:`database`), if provided via
the ``.env`` file. Defaults to None.
**kwargs: Pass-through keyword arguments.
Returns:
EagerResult: Query results object.
"""
return self._driver.execute_query(query_=query,
parameters_=parameters,
database_=database or self._db,
**kwargs)
[docs]
def load_from_csv(self,
*,
path_nodes: str=None,
path_edges: str=None) -> tuple[bool, list]:
"""Load graph data from a CSV file.
This function is designed to load node and/or edge (relationship)
data. The required format for each CSV type is as follows, where
the property data is *optional*, whereas the ``id`` and ``label``
are *required*.
**Node Data**
- **Field Definitions**
- **id:** Unique (user-defined) node ID.
- **label:** Node label. For example: Book, Character, etc.
- **Properties:** The column heading for each property will
be used as its key in the graph.
- **Format:**
``id,label,prop1,...,propN``
**Edge Data**
- **Field Definitions**
- **source:** The ID of the *source* node.
- **target:** The ID of the *target* node.
- **type:** edge type. For example: AUTHOR,
APPEARS_IN, etc.
- **Properties:** The column heading for each property will
be used as its key in the graph.
- **Format:**
``source,target,type,prop1,...,propN``
Args:
path_nodes (str, optional): Full path to the node input data.
Defaults to None.
path_edges (str, optional): Full path to the edge input data.
Defaults to None.
Returns:
tuple[bool, list]: A tuple containing a success flag and a
list of query result objects.
"""
n = []
e = []
src = None
if path_nodes:
n = utilities.csv_to_nodes(path=path_nodes)
src = os.path.basename(path_nodes)
if path_edges:
e = utilities.csv_to_edges(path=path_edges)
if src is None:
src = os.path.basename(path_edges)
go = GraphObject(nodes=n, edges=e, source=src)
return self._add_graph_object(graphobject=go)
def _add_graph_object(self, graphobject: GraphObject) -> tuple[bool, list]: # noqa
"""Add nodes and relationships to the graph via GraphObject.
Args:
graphobjects (GraphObject): A :class:`GraphObject`
container class wrapping the :class:`Node` and
:class:`Edge` objects to be added to the graph.
Returns:
tuple[bool, list]: A tuple containing a success flag and a
list of Neo4j result objects.
"""
results = []
for node in graphobject.nodes:
q = utilities.merge_node_query(node=node)
results.append(self.execute_query(q, parameters=node.properties))
for edge in graphobject.edges:
q = utilities.merge_edge_query(edge=edge)
results.append(self.execute_query(q, parameters=edge.properties))
act = len(results)
exp = graphobject.count
if exp == act:
print(f'{act} nodes/edges added successfully.')
else: # nocover
print('[ERROR]: {act} of {exp} nodes/edges added.')
return (exp == act), results
def _attempt_credentials(self) -> None:
"""Attempt to use the ``.env`` file for driver connection."""
if not all((os.path.isfile(self._auth), os.path.exists(self._auth))):
raise RuntimeError('The authentication file either is not a file or '
f'cannot be found: {self._auth}')
dotenv.load_dotenv(dotenv_path=self._auth)
if self._uri is None:
self._uri = os.environ.get('NEO4J_URI')
self._auth = (os.environ.get('NEO4J_USERNAME'), os.environ.get('NEO4J_PASSWORD'))
# Class constructor overrides the database name.
if self._db is None:
self._db = os.environ.get('NEO4J_DATABASE')
def _connect(self) -> None:
"""Connect to the database.
If the ``auth`` argument is a file path, an attempt is made to
use the credentials from the ``.env`` file. Otherwise, the
``auth`` argument credentials tuple is used.
"""
if self._auth is None: # nocover
self._auth = os.path.join(self._DIR_ROOT, '.env')
if isinstance(self._auth, str):
self._attempt_credentials()
self._driver = _GraphDatabase.driver(uri=self._uri, auth=self._auth, **self._config)
if self._driver is None or not self._driver.verify_authentication():
raise RuntimeError('Database authentication failed. Review the credentials.')
assert self._driver.verify_connectivity() is None
self._connected = True
[docs]
def create_dotenv_file(uri: str='neo4j://127.0.0.1:7687',
database: str='neo4j',
password: str='neo4j',
username: str='neo4j',
replace: bool=False) -> None: # nocover
"""Generate a ``.env`` file from the provided credentials.
The resulting file will be placed on the user's Desktop.
Args:
uri (str, optional): Database location URI.
Defaults to 'neo4j://127.0.0.1:7687'.
database (str, optional): Database name. Defaults to 'neo4j'.
password (str, optional): Authentication password.
Defaults to 'neo4j'.
username (str, optional): Authentication username.
Defaults to 'neo4j'.
replace (bool, optional): Force replace an existing ``.env``
file. Defaults to False.
Raises:
RuntimeError: Raised if a ``.env`` file already exists on the
user's Desktop *and* ``replace=False``.
"""
path = os.path.expanduser('~/Desktop/.env')
if os.path.exists(path) and not replace:
raise RuntimeError('An .env file already exists on the Desktop. '
'To overwrite, pass replace=True.')
with open(path, 'w', encoding='utf-8') as f:
f.write(f'NEO4J_URI="{uri}"\n')
f.write(f'NEO4J_DATABASE="{database}"\n')
f.write(f'NEO4J_PASSWORD="{password}"\n')
f.write(f'NEO4J_USERNAME="{username}"')