API Reference

Protocols

`GraphProtocol`

All operations on a single graph database.

Provides CRUD for nodes and edges, traversal queries, raw query passthrough, bulk operations, and schema management.

async def create_node(
    labels: list[str],
    properties: dict[str, Any] | None = None,
    node_id: str | None = None
) -> NodeResult

Create a node with labels and properties.

async def get_node(node_id: str) -> GraphNode | None

Retrieve a node by ID. Returns None if not found.

async def find_nodes(
    labels: list[str] | None = None,
    filter: PropertyFilter | None = None,
    limit: int = 100,
    skip: int = 0
) -> list[GraphNode]

Find nodes matching labels and/or property filter.

async def update_node(
    node_id: str,
    properties: dict[str, Any],
    merge: bool = True
) -> bool

Update node properties.

async def delete_node(
    node_id: str,
    detach: bool = True
) -> bool

Delete a node.

async def neighbors(
    node_id: str,
    depth: int = 1,
    direction: EdgeDirection = EdgeDirection.BOTH,
    edge_types: list[str] | None = None
) -> list[GraphNode]

Get neighbouring nodes reachable within the given depth.

async def count_nodes() -> int

Return the total number of nodes in the graph.

async def count_edges() -> int

Return the total number of edges in the graph.

async def get_labels() -> list[str]

Return all unique node labels in the graph.

async def get_edge_types() -> list[str]

Return all unique edge types in the graph.

async def create_edge(
    source_id: str,
    target_id: str,
    edge_type: str,
    properties: dict[str, Any] | None = None
) -> EdgeResult

Create a directed edge between two nodes.

async def get_edge(edge_id: str) -> GraphEdge | None

Retrieve an edge by ID.

async def get_edges(
    node_id: str,
    direction: EdgeDirection = EdgeDirection.BOTH,
    edge_types: list[str] | None = None,
    limit: int = 100
) -> list[GraphEdge]

Get edges connected to a node.

async def update_edge(
    edge_id: str,
    properties: dict[str, Any],
    merge: bool = True
) -> bool

Update edge properties.

async def delete_edge(edge_id: str) -> bool

Delete an edge by ID.

async def traverse(query: TraversalQuery) -> list[GraphPath]

Execute a traversal query and return matching paths.

async def shortest_path(
    from_id: str,
    to_id: str,
    max_depth: int = 10,
    edge_types: list[str] | None = None,
    direction: EdgeDirection = EdgeDirection.BOTH
) -> GraphPath | None

Find the shortest path between two nodes.

async def query(
    query_string: str,
    parameters: dict[str, Any] | None = None
) -> list[dict[str, Any]]

Execute a raw backend-specific query (Cypher, Gremlin, etc.).

async def bulk_create_nodes(nodes: list[NodeSpec]) -> BulkNodeResult

Create multiple nodes in a single operation.

async def bulk_create_edges(edges: list[EdgeSpec]) -> BulkEdgeResult

Create multiple edges in a single operation.

async def create_index(spec: IndexSpec) -> None

Create an index on node label + properties.

async def drop_index(name: str) -> None

Drop an index by name.

async def create_constraint(spec: ConstraintSpec) -> None

Create a schema constraint.

async def drop_constraint(name: str) -> None

Drop a constraint by name.

`GraphStoreProtocol`

Top-level graph store lifecycle and database management.

Manages connections to the graph backend and provides access to individual graph databases.

async def connect() -> None

Establish connection to the graph store.

async def disconnect() -> None

Close all connections and release resources.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check store connectivity and readiness.

async def get_graph(name: str | None = None) -> GraphProtocol

Get a handle to a graph database.

Parameters

Parameter	Type	Description
`name`	str \| None	Database name. ``None`` returns the default graph.

Raises

Exception	Description
GraphNotFoundError	If the named graph does not exist.

async def list_graphs() -> list[GraphInfo]

List all available graph databases.

async def create_graph(name: str) -> None

Create a new graph database.

Raises

Exception	Description
GraphAlreadyExistsError	If the graph already exists.

async def delete_graph(name: str) -> None

Delete a graph database and all its data.

Raises

Exception	Description
GraphNotFoundError	If the graph does not exist.

Classes

`BaseGraph`

Common logic for all graph database implementations.

def __init__(name: str | None = None) -> None

property name() -> str

Graph name.

`BaseGraphStore`

Common logic for all graph store implementations.

async def connect() -> None

Establish connection.

async def disconnect() -> None

Close connection.

`GraphConfig`

Top-level graph store configuration.

Loaded from the graph: key in application.yaml, with environment variable overrides via LEX_GRAPH__* prefix.

def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]

Check config is safe for the target environment.

Parameters

Parameter	Type	Description
`env`	Environment \| None	The target deployment environment. Defaults to the current environment.

Returns

Type	Description
list[ConfigIssue]	A list of ConfigIssue describing any configuration problems found.

`GraphConnectedEvent`

Raised when a graph database connection is established.

Downstream: observability, health monitoring.

`GraphDisconnectedEvent`

Raised when a graph database connection is closed.

Downstream: observability, cleanup hooks.

`GraphEdgeCreatedEvent`

Raised when an edge (relationship) is created between two nodes.

Downstream: audit logging, event sourcing.

`GraphModule`

Graph storage module — registers GraphStoreProtocol and GraphProtocol.

Registers GraphStoreProtocol and GraphProtocol for constructor injection.

Call configure to configure the graph backend (Neo4j or in-memory), or stub for an isolated setup with no external service dependencies.

Usage

from lexigram.graph.config import GraphConfig

@module(
    imports=[GraphModule.configure(GraphConfig(backend="neo4j"))]
)
class AppModule(Module):
    pass

def configure(
    cls,
    config: GraphConfig | None = None
) -> DynamicModule

Create a GraphModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	GraphConfig \| None	GraphConfig, a plain ``dict`` of config values, or ``None`` to use environment variable defaults.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

Raises

Exception	Description
TypeError	If config is not a ``GraphConfig``, ``dict``, or ``None``.

def stub(
    cls,
    config: GraphConfig | None = None
) -> DynamicModule

Create a GraphModule suitable for unit and integration testing.

Uses an in-memory backend with no external service dependencies.

Parameters

Parameter	Type	Description
`config`	GraphConfig \| None	Optional GraphConfig override. Uses safe in-memory defaults when ``None``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

`GraphNodeCreatedEvent`

Raised when a node is created in the graph.

Downstream: audit logging, event sourcing.

`GraphProvider`

def __init__(config: GraphConfig | None = None) -> None

async def register(container: ContainerRegistrarProtocol) -> None

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The DI registrar to register services into.

Raises

Exception	Description
ValueError	If an unknown backend is specified in config.

async def boot(container: ContainerResolverProtocol) -> None

Connect to the graph store after the container is fully built.

Parameters

Parameter	Type	Description
`container`	ContainerResolverProtocol	The DI resolver to obtain the graph store from.

async def shutdown() -> None

Disconnect from the graph store during application shutdown.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check provider health.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for health check response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with status and component details.

`GraphQueryExecutedEvent`

Raised when a graph query is executed (e.g., Cypher).

Downstream: analytics, slow query logging.

`Neo4jConfig`

Configuration for the Neo4j backend.

`Neo4jGraph`

Neo4j graph implementation (Cypher-based).

def __init__(
    driver: AsyncDriver,
    name: str | None = None
) -> None

async def create_node(
    labels: list[str],
    properties: dict[str, Any] | None = None,
    node_id: str | None = None
) -> NodeResult

Create a node in Neo4j.

Parameters

Parameter	Type	Description
`labels`	list[str]	Node labels.
`properties`	dict[str, Any] \| None	Optional property map.
`node_id`	str \| None	Optional explicit node ID; stored as the ``id`` property.

Returns

Type	Description
NodeResult	NodeResult with the node ID.

async def get_node(node_id: str) -> GraphNode | None

Retrieve a node by its id property.

Parameters

Parameter	Type	Description
`node_id`	str	The node's ``id`` property value.

Returns

Type	Description
GraphNode \| None	The GraphNode or ``None``.

async def find_nodes(
    labels: list[str] | None = None,
    filter: PropertyFilter | None = None,
    limit: int = 100,
    skip: int = 0
) -> list[GraphNode]

Find nodes with optional label filtering.

Parameters

Parameter	Type	Description
`labels`	list[str] \| None	Optional label filter; all labels must match.
`filter`	PropertyFilter \| None	Unused.
`limit`	int	Maximum results.
`skip`	int	Results to skip.

Returns

Type	Description
list[GraphNode]	Matching GraphNode instances.

async def create_edge(
    source_id: str,
    target_id: str,
    edge_type: str,
    properties: dict[str, Any] | None = None
) -> EdgeResult

Create a directed relationship between two nodes.

Parameters

Parameter	Type	Description
`source_id`	str	Source node ``id`` property.
`target_id`	str	Target node ``id`` property.
`edge_type`	str	Relationship type.
`properties`	dict[str, Any] \| None	Optional relationship properties.

Returns

Type	Description
EdgeResult	EdgeResult with the element ID of the new relationship.

async def traverse(query: TraversalQuery) -> list[GraphPath]

Execute a traversal query via compiled Cypher.

Parameters

Parameter	Type	Description
`query`	TraversalQuery	Traversal parameters including start node, depth, and relationship type filters.

Returns

Type	Description
list[GraphPath]	All GraphPath instances found by the traversal.

async def query(
    query_string: str,
    parameters: dict[str, Any] | None = None
) -> list[dict[str, Any]]

Execute a raw Cypher query string.

Parameters

Parameter	Type	Description
`query_string`	str	Cypher query.
`parameters`	dict[str, Any] \| None	Optional parameter bindings.

Returns

Type	Description
list[dict[str, Any]]	List of record dicts.

`Neo4jGraphStore`

Neo4j managed graph store backend.

def __init__(config: Neo4jConfig) -> None

async def connect() -> None

Connect to Neo4j and create foundational constraints.

Creates a uniqueness constraint for the id property on Node labels as a best-effort operation.

async def disconnect() -> None

Close the Neo4j driver.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Verify connectivity to Neo4j.

Parameters

Parameter	Type	Description
`timeout`	float	Unused; Neo4j driver manages its own timeout.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with HEALTHY or UNHEALTHY status.

async def get_graph(name: str | None = None) -> Neo4jGraph

Return a handle to a named Neo4j database.

Parameters

Parameter	Type	Description
`name`	str \| None	Database name. Defaults to the driver default.

Returns

Type	Description
Neo4jGraph	A Neo4jGraph instance scoped to the given database.

Raises

Exception	Description
RuntimeError	If not yet connected.

async def list_graphs() -> list[GraphInfo]

List all databases visible to the connected Neo4j instance.

Returns

Type	Description
list[GraphInfo]	A list of GraphInfo.

async def query(query_string: str) -> list[dict[str, Any]]

Execute a raw Cypher query and return all records.

Parameters

Parameter	Type	Description
`query_string`	str	Cypher query to execute.

Returns

Type	Description
list[dict[str, Any]]	List of record dicts.

Raises

Exception	Description
RuntimeError	If not yet connected.