🔍 Code Extractor

A manager class that provides a high-level interface for interacting with Neo4j graph databases, handling connections, queries, node creation, and relationship management.

Purpose

Neo4jManager encapsulates all Neo4j database operations, providing methods to execute Cypher queries, create nodes with labels and properties, and establish relationships between nodes. It manages the database connection lifecycle using the Neo4j Python driver and provides error handling and logging for all database operations. This class is designed to be instantiated once per application or service and reused for all Neo4j interactions.

Source Code

class Neo4jManager:
    """Manages interactions with the Neo4j graph database."""

    def __init__(self, config: ConfigParser):
        """
        Initializes the Neo4jManager with the database URI and credentials specified in the config.

        Args:
            config (ConfigParser): Configuration object containing database settings.
        """
        self.logger = logging.getLogger(__name__)
        self.uri = config.get('database', 'neo4j_uri', fallback='bolt://localhost:7687')
        self.username = config.get('database', 'neo4j_username', fallback='neo4j')
        self.password = config.get('database', 'neo4j_password', fallback='password')
        self.driver = GraphDatabase.driver(self.uri, auth=(self.username, self.password))

        self.logger.info(f"Neo4jManager initialized with URI: {self.uri}")

    def execute_query(self, query: str, parameters: Dict[str, Any] = None) -> List[Dict[str, Any]]:
        """
        Executes a Cypher query against the Neo4j database.

        Args:
            query (str): The Cypher query to execute.
            parameters (Dict[str, Any]], optional): A dictionary of parameters to pass to the query. Defaults to None.

        Returns:
            List[Dict[str, Any]]: A list of dictionaries representing the results of the query.
        """
        try:
            with self.driver.session() as session:
                result = session.run(query, parameters)
                records = [record.data() for record in result]
                self.logger.info(f"Executed query: {query}")
                return records
        except Exception as e:
            self.logger.exception(f"Error executing Neo4j query: {e}")
            return []

    def create_node(self, label: str, properties: Dict[str, Any]) -> None:
        """
        Creates a node in the Neo4j database with the given label and properties.

        Args:
            label (str): The label for the node.
            properties (Dict[str, Any]): A dictionary of properties for the node.
        """
        try:
            query = f"CREATE (n:{label} $properties)"
            self.execute_query(query, {"properties": properties})
            self.logger.info(f"Created node with label '{label}' and properties: {properties}")
        except Exception as e:
            self.logger.exception(f"Error creating Neo4j node: {e}")

    def create_relationship(self, node1_label: str, node1_property: str, node1_value: Any,
                              relationship_type: str,
                              node2_label: str, node2_property: str, node2_value: Any,
                              relationship_properties: Dict[str, Any] = None) -> None:
        """
        Creates a relationship between two nodes in the Neo4j database.

        Args:
            node1_label (str): The label for the first node.
            node1_property (str): The property to match for the first node.
            node1_value (Any): The value of the property to match for the first node.
            relationship_type (str): The type of relationship to create.
            node2_label (str): The label for the second node.
            node2_property (str): The property to match for the second node.
            node2_value (Any): The value of the property to match for the second node.
            relationship_properties (Dict[str, Any]], optional): A dictionary of properties for the relationship. Defaults to None.
        """
        try:
            query = (
                f"MATCH (n1:{node1_label} {{{node1_property}: $node1_value}}), "
                f"(n2:{node2_label} {{{node2_property}: $node2_value}}) "
                f"CREATE (n1)-[r:{relationship_type} $relationship_properties]->(n2)"
            )
            parameters = {
                "node1_value": node1_value,
                "node2_value": node2_value,
                "relationship_properties": relationship_properties if relationship_properties else {}
            }
            self.execute_query(query, parameters)
            self.logger.info(f"Created relationship '{relationship_type}' between nodes: "
                             f"({node1_label}:{node1_property}={node1_value}) and "
                             f"({node2_label}:{node2_property}={node2_value})")
        except Exception as e:
            self.logger.exception(f"Error creating Neo4j relationship: {e}")

Parameters

Parameter Details

config: A ConfigParser object containing database configuration settings. Must include 'database' section with optional keys: 'neo4j_uri' (defaults to 'bolt://localhost:7687'), 'neo4j_username' (defaults to 'neo4j'), and 'neo4j_password' (defaults to 'password'). This configuration object is used to establish the database connection.

Return Value

Instantiation returns a Neo4jManager object with an active database driver connection. The execute_query method returns a List[Dict[str, Any]] containing query results as dictionaries. The create_node and create_relationship methods return None but perform side effects on the database.

Class Interface

Methods

Attributes

Dependencies

• neo4j
• logging
• typing
• configparser

Required Imports

from neo4j import GraphDatabase
import logging
from typing import List, Dict, Any
from configparser import ConfigParser

Usage Example

from configparser import ConfigParser
from neo4j import GraphDatabase
import logging
from typing import List, Dict, Any

# Setup configuration
config = ConfigParser()
config.add_section('database')
config.set('database', 'neo4j_uri', 'bolt://localhost:7687')
config.set('database', 'neo4j_username', 'neo4j')
config.set('database', 'neo4j_password', 'mypassword')

# Initialize manager
manager = Neo4jManager(config)

# Create nodes
manager.create_node('Person', {'name': 'Alice', 'age': 30})
manager.create_node('Person', {'name': 'Bob', 'age': 25})

# Create relationship
manager.create_relationship(
    node1_label='Person',
    node1_property='name',
    node1_value='Alice',
    relationship_type='KNOWS',
    node2_label='Person',
    node2_property='name',
    node2_value='Bob',
    relationship_properties={'since': 2020}
)

# Execute custom query
results = manager.execute_query(
    'MATCH (p:Person) WHERE p.age > $min_age RETURN p.name as name, p.age as age',
    {'min_age': 20}
)
for record in results:
    print(f"Name: {record['name']}, Age: {record['age']}")

# Close driver when done
manager.driver.close()

Best Practices

• Always close the driver connection when done using manager.driver.close() to release resources
• Reuse a single Neo4jManager instance throughout your application rather than creating multiple instances
• Ensure the Neo4j database is running and accessible before instantiating the manager
• Use parameterized queries through the parameters argument to prevent Cypher injection attacks
• Handle empty result lists from execute_query as it returns [] on errors
• Check logs for detailed error information as exceptions are caught and logged but not re-raised
• Ensure nodes exist before creating relationships between them, or the relationship creation will silently fail
• Use appropriate node labels and property names that follow Neo4j naming conventions
• Consider implementing connection pooling for high-throughput applications
• The driver maintains its own connection pool, so session creation is lightweight

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class Graph 74.9% similar

  A Graph class that provides an interface for interacting with a Neo4j graph database, supporting CRUD operations on nodes and relationships through Cypher queries.

  From: /tf/active/vicechatdev/neo4j_driver/neo4j_driver.py

• class Node 62.6% similar

  A Node class representing a graph node with labels and properties, designed to work with Neo4j graph databases. It extends PropertyDict to manage node properties and provides methods for label management and graph synchronization.

  From: /tf/active/vicechatdev/neo4j_driver/neo4j_objects.py

• class ManualRelationshipManager 58.7% similar

  A class that manages manually defined database relationships with persistent JSON storage, allowing users to add, retrieve, update, and remove relationship definitions between database tables.

  From: /tf/active/vicechatdev/full_smartstat/manual_relationships.py

• function init_connections 58.2% similar

  Initializes and returns a Neo4j database session and driver connection using configuration settings.

  From: /tf/active/vicechatdev/offline_docstore_multi_vice.py

• class Relationship 57.4% similar

  A class representing a graph relationship between two nodes, wrapping a _Relationship object with start and end Node objects.

  From: /tf/active/vicechatdev/neo4j_driver/neo4j_objects.py