Graph Database (Neo4j)

The athomic.database.graph module provides a high-performance, asynchronous interface for interacting with Graph Databases. Currently, it ships with a robust Neo4j implementation.

It is designed to support complex relationship mapping, such as Data Lineage, Impact Analysis, and Knowledge Graphs.

Features

• Async-First: Built on top of the official neo4j asynchronous driver.
• Observability: Automatic OpenTelemetry tracing (spans) and Prometheus metrics for every query executed.
• Connection Groups: Supports multiple named connections (e.g., default, analytics) via ConnectionManager.
• Resilience: Integrated connection pooling and automatic retry mechanisms managed by the driver.

---

Configuration

Configure the graph connection in your settings.toml. The module uses the standard ConnectionGroup pattern found in other database modules.

[default.database.graph]
# The default connection to use when none is specified
default_connection_name = "default_graph"

[default.database.graph.connections.default_graph]
enabled = true
backend = "neo4j"

[default.database.graph.connections.default_graph.provider]
backend = "neo4j"
# Connection URI (supports bolt://, neo4j://, neo4j+s://)
uri = "bolt://localhost:7687"
user = "neo4j"
password = "password" # pragma: allowlist secret
max_connection_pool_size = 50
max_connection_lifetime = 3600

---

Usage

Dependency Injection (Recommended)

The best way to interact with the Graph DB is by injecting the GraphDatabaseProtocol via the ConnectionManager or the specific Factory.

from nala.athomic.database.graph.protocol import GraphDatabaseProtocol
from nala.athomic.database.factory import connection_manager_factory

async def get_service_dependencies():
    # 1. Get the manager
    connection_manager = connection_manager_factory.create()

    # 2. Get the client (uses 'default_connection_name' if name is omitted)
    graph_client: GraphDatabaseProtocol = connection_manager.get_graph_db()

    # 3. Execute a Cypher query
    query = """
    MATCH (s:Service)-[:DEPENDS_ON]->(d:Database)
    WHERE s.name = $service_name
    RETURN d.name as database, d.type as type
    """

    results = await graph_client.execute_query(
        query, 
        parameters={"service_name": "payment-service"}
    )

    for record in results:
        print(f"Database: {record['database']}, Type: {record['type']}")

Multiple Connections

If you have configured multiple connections (e.g., one for Lineage, one for Fraud Detection), you can request them by name:

# Accessing the 'fraud_graph' connection configured in settings.toml
fraud_graph = connection_manager.get_graph_db("fraud_graph")

---

Architecture & Patterns

The Protocol

The module exposes GraphDatabaseProtocol. Your domain logic should depend on this protocol, not on Neo4jProvider directly. This ensures strict Dependency Inversion (DIP).

Observability

You do not need to add manual logging or timing code. The BaseGraphStore wrapper automatically:

1. Starts a Trace Span named graph_db.execute_query.
2. Records Prometheus Metrics:
   • nala_graph_db_operations_total{status="success|failure"}
   • nala_graph_db_operation_duration_seconds

Error Handling

Driver-specific exceptions (like neo4j.exceptions.Neo4jError) are logged and propagated. In infrastructure layers, we wrap critical connection errors in ServiceConnectionError to standardize health checks.