Graph Databases

Overview

The athomic.database.graph module provides the abstraction layer for connecting to and managing graph-oriented databases within the Athomic Layer. It allows the application to model complex relationships, such as data lineage, social networks, or impact analysis graphs, using a high-performance, asynchronous interface.

With the introduction of Wrappers, this module now supports transparent, query-level Multi-Tenancy, ensuring that data isolation is enforced automatically without cluttering business logic.

Core Concepts

GraphDatabaseProtocol

This is the abstract contract that all graph database providers must implement. It defines the standard interface for executing queries (run_query) and managing the connection lifecycle.

GraphManager

The GraphManager is the orchestrator for graph connections. It is responsible for: - Lifecycle: Ensuring that all configured graph drivers are connected at startup and gracefully closed at shutdown. - Registry: Providing access to specific client instances via get_client("connection_name").

GraphDatabaseFactory & Wrappers

The factory is responsible for instantiating the concrete provider (e.g., Neo4jProvider) and dynamically applying a chain of Wrappers (Middleware) based on configuration.

• Neo4jProvider: The base driver handling the raw connection to Neo4j.
• ContextAwareGraphClient (Wrapper): A decorator that automatically intercepts every query to inject the current tenant_id (from the Context module) into the query parameters. This ensures that every operation is scoped to the correct tenant.

---

How It Works

1. Configuration: You define your graph connections in settings.toml under the [database.graph] section.
2. Initialization: At application startup, the central ConnectionManager initializes the GraphManager.
3. Factory Creation: The GraphManager uses the GraphDatabaseFactory to instantiate the configured backend and wrap it with configured middlewares.
4. Usage: Services inject the GraphDatabaseProtocol via the connection manager to execute queries without knowing the implementation details.

---

Usage Example

Basic Query Execution

Services interact with the graph database via the GraphDatabaseProtocol. Thanks to the wrapper, you do not need to manually pass the tenant_id.

from nala.athomic.database.factory import connection_manager_factory

async def get_service_dependencies(service_name: str):
    # 1. Get the connection manager (Singleton)
    manager = connection_manager_factory.create()

    # 2. Get the default graph client
    # This instance is likely a 'ContextAwareGraphClient' wrapping 'Neo4jProvider'
    graph_db = manager.get_graph_db() 

    # 3. Define a Cypher query
    # Note: We use '$tenant_id' in the query, but we don't pass it in parameters.
    # The wrapper injects it automatically from the context.
    query = """
    MATCH (s:Service {name: $name, tenant_id: $tenant_id})-[:CALLS]->(target:Service)
    RETURN target.name as dependency
    """

    # 4. Execute asynchronously
    results = await graph_db.run_query(
        query, 
        parameters={"name": service_name} # 'tenant_id' is added automatically
    )

    return [record["dependency"] for record in results]

---

Configuration

Graph connections are configured using the standard Connection Group pattern in settings.toml. You can now define wrappers to enable features like multi-tenancy.

[default.database.graph]
# The name of the default connection to return when none is specified.
default_connection_name = "default_graph"

  # --- Connection 1: The main application graph ---
  [default.database.graph.connections.default_graph]
  enabled = true
  backend = "neo4j"

    # --- Provider Settings ---
    [default.database.graph.connections.default_graph.provider]
    backend = "neo4j"
    uri = "bolt://localhost:7687"
    user = "neo4j"
    # Secure password reference via the Secrets module
    password = { path = "database/neo4j", key = "password" } # pragma: allowlist secret

    # Connection pool tuning
    max_connection_pool_size = 50
    max_connection_lifetime = 3600

    # --- Wrapper Configuration ---
    # Wrappers apply cross-cutting concerns like Multi-Tenancy
    [[default.database.graph.connections.default_graph.wrappers]]
    name = "context_aware"
    enabled = true

---

API Reference

nala.athomic.database.graph.protocol.GraphDatabaseProtocol

Bases: BaseServiceProtocol, Protocol

Defines the contract for a Graph Database provider. This ensures the application is decoupled from specific graph technology drivers.

get_driver() async

Returns the underlying native driver instance.

run_query(query, parameters=None) async

Executes a Cypher (or equivalent) query and returns a list of records.

Parameters:

Name	Type	Description	Default
query	str	The query string.	required
parameters	Optional[Dict[str, Any]]	Dictionary of parameters to bind to the query.	None

Returns:

Type	Description
List[Dict[str, Any]]	List of dictionaries representing the result records.

nala.athomic.database.graph.manager.GraphManager

Bases: BaseManager[GraphDatabaseProtocol, GraphSettings]

A specialized lifecycle manager for Graph Database connections.

This class orchestrates the initialization, connection, and shutdown of multiple graph database providers (e.g., 'default', 'analytics'). It acts as the single point of access for retrieving specific graph clients within the application.

It inherits robust lifecycle management from BaseManager, ensuring that all configured graph connections are established during startup and gracefully closed during shutdown.

__init__(settings)

Initializes the GraphManager.

Parameters:

Name	Type	Description	Default
settings	DatabaseSettings	The root database settings object containing the 'graph' connection group configuration.	required

nala.athomic.database.graph.providers.neo4j.Neo4jProvider

Bases: GraphDatabaseBase, CredentialResolve

Concrete implementation of the GraphDatabaseProtocol for Neo4j.

This provider manages the asynchronous connection lifecycle to a Neo4j database and executes Cypher queries. It utilizes the official neo4j Python driver and handles credential resolution securely.

__init__(settings)

Initializes the Neo4jProvider instance.

Parameters:

Name	Type	Description	Default
settings	GraphSettings	The root database settings object containing the 'graph' connection group configuration.	required

Raises:

Type	Description
ProviderInitializationError	If the provider configuration is not of type Neo4jSettings.

get_driver() async

Returns the underlying Neo4j driver instance.

This is useful for advanced operations (e.g. transactions, migrations) that require direct access to the driver.

Returns:

Name	Type	Description
Any	Any	The neo4j.AsyncDriver instance.

run_query(query, parameters=None) async

Executes a Cypher query against the graph database.

This is the public entry point required by the GraphDatabaseProtocol.

Parameters:

Name	Type	Description	Default
query	str	The Cypher query string to execute.	required
parameters	Dict[str, Any] | None	A dictionary of query parameters.	None

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: A list of records, where each record is a dictionary.

Raises:

Type	Description
ServiceConnectionError	If the driver is not connected.

nala.athomic.database.graph.wrappers.context_aware.ContextAwareGraphClient

Bases: GraphWrapperBase

Wrapper that injects multitenancy context into graph database operations.

This client automatically retrieves the current tenant_id (and potentially user_id) from the application context and injects them as parameters into every Cypher query execution. This ensures that data isolation policies can be enforced at the query level without manual parameter passing in business logic.

__init__(client)

Initializes the context-aware wrapper.

Parameters:

Name	Type	Description	Default
client	GraphDatabaseProtocol	The underlying GraphDatabaseProtocol instance.	required

run_query(query, parameters=None) async

Executes a Cypher query with context parameters automatically injected.

Parameters:

Name	Type	Description	Default
query	str	The Cypher query string.	required
parameters	Optional[Dict[str, Any]]	Optional query parameters provided by the caller.	None

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: The query results.

nala.athomic.database.graph.factory.GraphDatabaseFactory

Factory for creating GraphDatabaseProtocol instances. Responsible for instantiating the provider and applying configured wrappers.

create(settings=None) classmethod

Creates a graph provider instance (potentially wrapped).

If 'settings' is NOT provided, it resolves the DEFAULT connection from the global AppSettings.