Class DatabaseManager

src/codegraphcontext/core/database.py:50–350

Manages the Neo4j database driver as a singleton to ensure only one connection pool is created and shared across the application. This pattern is crucial for performance and resource management in a multi-threaded or asynchronous application.

        self._driver.close()

class DatabaseManager:
    """
    Manages the Neo4j database driver as a singleton to ensure only one
    connection pool is created and shared across the application.
    
    This pattern is crucial for performance and resource management in a
    multi-threaded or asynchronous application.
    """
    _instance = None
    _driver: Optional[Driver] = None
    _lock = threading.Lock() # Lock to ensure thread-safe initialization. 

    def __new__(cls):
        """Standard singleton pattern implementation."""
        if cls._instance is None:
            with cls._lock:
                # Double-check locking to prevent race conditions.
                if cls._instance is None:
                    cls._instance = super(DatabaseManager, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        """
        Initializes the manager by reading credentials from environment variables.
        The `_initialized` flag prevents re-initialization on subsequent calls.
        """
        if hasattr(self, '_initialized'):
            return

        self.neo4j_uri = os.getenv('NEO4J_URI')
        self.neo4j_username = os.getenv('NEO4J_USERNAME', 'neo4j')
        self.neo4j_password = os.getenv('NEO4J_PASSWORD')
        self.neo4j_database = os.getenv('NEO4J_DATABASE') # Optional, if not set, will use default database configured in Neo4j
        self._initialized = True

    def get_driver(self, graph_name: str = None) -> Driver:
        """
        Gets the Neo4j driver instance, creating it if it doesn't exist.
        This method is thread-safe.

        The `graph_name` parameter is accepted for interface parity with
        FalkorDB (which supports multiple graphs per instance); Neo4j
        selects the database via NEO4J_DATABASE, so the argument is ignored.

        Raises:
            ValueError: If Neo4j credentials are not set in environment variables.

        Returns:
            The a wrapper for Neo4j Driver instance.
        """
        if self._driver is None:
            with self._lock:
                if self._driver is None:
                    # Ensure all necessary credentials are provided.
                    missing = self.get_missing_credentials(
                        self.neo4j_uri,
                        self.neo4j_username,
                        self.neo4j_password,