grafeo.GrafeoDB¶

The main database class.

Constructor¶

GrafeoDB(
    path: Optional[str] = None,
    *,
    cdc: bool = False,
    shuffle_unordered: bool = False,
    wal_salvage: bool = False,
    history_retention: Optional[Union[str, timedelta]] = None
)

Parameters¶

Examples¶

# In-memory database
db = grafeo.GrafeoDB()

# Persistent database
db = grafeo.GrafeoDB("my_graph.db")

Static Constructors¶

open()¶

Open an existing database.

@staticmethod
def open(path: str) -> GrafeoDB

open_read_only()¶

Open a database in read-only mode. Uses a shared file lock, so multiple processes can read the same database file concurrently. Mutations will raise an error. A database written by 0.5.x (a file or a WAL directory) is read into memory once, with no lock held afterwards, and is not migrated.

@staticmethod
def open_read_only(path: str) -> GrafeoDB

db = GrafeoDB.open_read_only("./my_graph.grafeo")
result = db.execute("MATCH (n) RETURN n LIMIT 10")

open_in_memory()¶

Open a persistent database file and load it entirely into memory. The returned database has no connection to the original file: changes will not be written back.

@staticmethod
def open_in_memory(path: str) -> GrafeoDB

db = GrafeoDB.open_in_memory("./mydb")
db.create_node(["Test"], {})  # does not affect the file

Query Methods¶

execute()¶

Execute a GQL query.

def execute(self, query: str, params: Optional[Dict] = None) -> QueryResult

execute_cypher()¶

Execute a Cypher query.

def execute_cypher(self, query: str, params: Optional[Dict] = None) -> QueryResult

execute_gremlin()¶

Execute a Gremlin query.

def execute_gremlin(self, query: str, params: Optional[Dict] = None) -> QueryResult

execute_graphql()¶

Execute a GraphQL query.

def execute_graphql(self, query: str, params: Optional[Dict] = None) -> QueryResult

execute_sparql()¶

Execute a SPARQL query.

def execute_sparql(self, query: str, params: Optional[Dict] = None) -> QueryResult

explain_sparql()¶

Execute a SPARQL query with EXPLAIN or EXPLAIN ANALYZE to inspect the physical plan tree. Returns a QueryResult containing the query plan as text rows. Use EXPLAIN for the plan without execution, or EXPLAIN ANALYZE for a profiled execution that includes timing and row counts.

def explain_sparql(self, query: str) -> QueryResult

# View the query plan without executing
result = db.explain_sparql("EXPLAIN SELECT ?name WHERE { ?p a <http://example.org/Person> ; <http://example.org/name> ?name }")
for row in result:
    print(row)

# Profiled execution with timing
result = db.explain_sparql("EXPLAIN ANALYZE SELECT ?s ?p ?o WHERE { ?s ?p ?o } LIMIT 100")
for row in result:
    print(row)

execute_sql()¶

Execute a SQL/PGQ query.

def execute_sql(self, query: str, params: Optional[Dict] = None) -> QueryResult

execute_async()¶

Execute a GQL query asynchronously. Returns a Python awaitable for use with asyncio.

def execute_async(self, query: str, params: Optional[Dict] = None) -> Awaitable[AsyncQueryResult]

import asyncio

async def main():
    db = GrafeoDB()
    result = await db.execute_async("MATCH (n:Person) RETURN n")
    for row in result:
        print(row)

asyncio.run(main())

execute_at_epoch()¶

Execute a GQL query at a specific historical epoch. Returns results as they would have appeared at that point in time.

def execute_at_epoch(self, query: str, epoch: int, params: Optional[Dict] = None) -> QueryResult

result = db.execute_at_epoch("MATCH (n:Server) RETURN n.status", epoch=5)

as_of()¶

A read-only view of the database at a point in time: an int (an epoch), a timezone-aware datetime (the last commit at or before it) or a timedelta (that long ago). A naive datetime raises ValueError, another type TypeError; a point before the history the database keeps raises GrafeoError ("not retained"). The view keeps the history of its epoch until it is closed: use it in a with block or call close().

def as_of(self, point: Union[int, datetime, timedelta]) -> AsOfView

with db.as_of(timedelta(minutes=5)) as view:
    print(view.epoch, view.execute("MATCH (p:Person) RETURN count(p)").scalar())

An AsOfView has epoch, execute(query, params=None), get_node(id), get_edge(id), get_node_property(id, key) and close(). A query that writes through it raises GrafeoError.

execute_as_of()¶

Execute one GQL query at a point in time (as for as_of()), without keeping a view.

def execute_as_of(self, query: str, point: Union[int, datetime, timedelta], params: Optional[Dict] = None) -> QueryResult

epoch_as_of()¶

The epoch a point in time reads at: the epoch itself (the current one for a later one), or the last commit at or before a time.

def epoch_as_of(self, point: Union[int, datetime, timedelta]) -> int

Node Operations¶

create_node()¶

Create a node with labels and properties.

def create_node(self, labels: List[str], properties: Optional[Dict[str, Any]] = None) -> Node

get_node()¶

Get a node by ID. Returns None if the node does not exist.

def get_node(self, id: int) -> Optional[Node]

delete_node()¶

Delete a node by ID. Returns True if the node existed and was deleted.

def delete_node(self, id: int) -> bool

A node that still has edges raises: delete them first, or use detach_delete_node().

detach_delete_node()¶

Delete a node with all its edges, as DETACH DELETE does. Returns True if the node existed. The node and its edges are one change: they commit and roll back together, and change data capture names the node in each edge's deleted_with.

def detach_delete_node(self, id: int) -> bool

add_node_label()¶

Add a label to an existing node. Returns True if the label was added, False if the node does not exist or already has the label.

def add_node_label(self, node_id: int, label: str) -> bool

remove_node_label()¶

Remove a label from a node. Returns True if the label was removed, False if the node does not exist or does not have the label.

def remove_node_label(self, node_id: int, label: str) -> bool

get_node_labels()¶

Get all labels for a node. Returns None if the node does not exist.

def get_node_labels(self, node_id: int) -> Optional[List[str]]

set_node_property()¶

Set a property on a node.

def set_node_property(self, node_id: int, key: str, value: Any) -> None

remove_node_property()¶

Remove a property from a node. Returns True if the property existed and was removed.

def remove_node_property(self, node_id: int, key: str) -> bool

get_nodes_by_label()¶

Get all nodes with a specific label and their properties. Supports pagination with limit and offset. More efficient than calling get_node() in a loop because it batches property lookups.

def get_nodes_by_label(
    self,
    label: str,
    limit: Optional[int] = None,
    offset: int = 0
) -> List[Tuple[int, Dict[str, Any]]]

Returns a list of (node_id, properties_dict) tuples.

people = db.get_nodes_by_label("Person", limit=100)
for node_id, props in people:
    print(f"Node {node_id}: {props}")

# Pagination
page = db.get_nodes_by_label("Person", limit=50, offset=100)

get_property_batch()¶

Get a specific property value for multiple nodes at once. More efficient than calling get_node() in a loop when you only need one property.

def get_property_batch(self, node_ids: List[int], property: str) -> List[Optional[Any]]

ages = db.get_property_batch([1, 2, 3, 4, 5], "age")
for node_id, age in zip([1, 2, 3, 4, 5], ages):
    if age is not None:
        print(f"Node {node_id} is {age} years old")

get_properties_batch()¶

Get several properties for multiple nodes in one call. Returns a dict with one list per key, in the order of node_ids: None where a node lacks the property or no node has the ID. IDs may repeat. No IDs gives an empty list per key, no keys gives an empty dict. It needs no pyarrow.

def get_properties_batch(self, node_ids: List[int], keys: List[str]) -> Dict[str, List[Optional[Any]]]

alix = db.create_node(["Person"], {"name": "Alix", "city": "Amsterdam"})
gus = db.create_node(["Person"], {"name": "Gus"})
columns = db.get_properties_batch([alix.id, gus.id], ["name", "city"])
# {'name': ['Alix', 'Gus'], 'city': ['Amsterdam', None]}

Edge Operations¶

create_edge()¶

Create an edge between two nodes.

def create_edge(
    self,
    source_id: int,
    target_id: int,
    edge_type: str,
    properties: Optional[Dict[str, Any]] = None
) -> Edge

get_edge()¶

Get an edge by ID. Returns None if the edge does not exist.

def get_edge(self, id: int) -> Optional[Edge]

get_node_edges()¶

The edges of a node as (neighbor_id, edge_id) tuples, without a query. direction is "out" (the default), "in" or "both", which lists a self-loop once; edge_types keeps only edges of those types (a type or a list of types). A node that does not exist has none. A graph handle (db.graph(name)) has the same method for its graph.

def get_node_edges(
    self,
    node_id: int,
    direction: str = "out",
    edge_types: Optional[Union[str, Sequence[str]]] = None,
) -> List[Tuple[int, int]]

for neighbor_id, edge_id in db.get_node_edges(alix.id, "out", ["KNOWS"]):
    print(db.get_node(neighbor_id))

delete_edge()¶

Delete an edge by ID. Returns True if the edge existed and was deleted.

def delete_edge(self, id: int) -> bool

set_edge_property()¶

Set a property on an edge.

def set_edge_property(self, edge_id: int, key: str, value: Any) -> None

remove_edge_property()¶

Remove a property from an edge. Returns True if the property existed and was removed.

def remove_edge_property(self, edge_id: int, key: str) -> bool

Upserts¶

Create or update nodes and edges by a key property, many rows in one statement, or update existing nodes only. The rows are checked like a query (constraints, schema), and a call writes all of its rows or none. In an upsert rows apply in order: a key repeated within one call creates one node or edge, which the later rows update. The upserts return a dict with created, updated, skipped and skipped_rows (the indices of the skipped rows, at most 1,000), update_nodes the same without created; the dicts gain keys only, as the admin dicts do. Graph handles (db.graph(name)) have the same methods.

upsert_nodes()¶

One node per row, matched by key and all of labels. A row without the key is skipped. By default a row's properties are merged into the node's and none is removed; with replace=True they become exactly the row's. Labels are never removed.

def upsert_nodes(self, labels: list[str], rows: list[dict], key: str = "id", replace: bool = False) -> dict

db.upsert_nodes(["Graph", "File"], [{"id": "f1", "size": 3}, {"id": "f2", "size": 5}])
# {'created': 2, 'updated': 0, 'skipped': 0, 'skipped_rows': []}

update_nodes()¶

Updates the nodes that already exist, one per row, matched by key and all of labels; it never creates a node. A row's properties are merged into the node's and the others stay, as UNWIND $rows AS r MATCH (n:Labels {key: r.key}) SET n += r does:

• a None value removes the property;
• a key that several nodes have updates all of them (the row counts once in updated);
• a row whose key matches no node with every label, or that lacks the key, is skipped and creates nothing;
• keys compare as = does, which also takes a string of digits as equal to that number: 2.0 and "2" find the node whose key is 2, and 3 the one whose key is "3";
• a row with only the key counts as updated and changes nothing;
• with no labels ([]), every node counts.

The call is one write, as that query is: all rows or none, logged to the WAL (a persistent database keeps it after a crash) and to change data capture, and inside the open transaction when there is one. No rows is a no-op. Without a property index on key it reads the nodes of the first label once, so its time grows with the rows and those nodes; with one, rows up to a third of those nodes look their nodes up through the index, as the query does.

def update_nodes(self, labels: list[str], rows: list[dict], key: str = "id") -> dict

db.upsert_nodes(["Person"], [{"id": "p1", "name": "Alix"}, {"id": "p2", "name": "Gus"}])
db.update_nodes(
    ["Person"],
    [{"id": "p1", "city": "Amsterdam"}, {"id": "p2", "city": "Berlin"}, {"id": "p3", "city": "Berlin"}],
)
# {'updated': 2, 'skipped': 1, 'skipped_rows': [2]}

upsert_edges()¶

One edge of edge_type per row, between the nodes whose endpoint_key is the row's src_field and dst_field value (restricted to endpoint_labels when given). The edge is identified by its endpoints, type and key; every other field of the row is an edge property. A row is skipped when it lacks the key, src_field or dst_field, or when no node or more than one node has its endpoint key; endpoints are never created. key, src_field and dst_field must be different fields. A property index on endpoint_key makes the lookups fast.

def upsert_edges(
    self,
    edge_type: str,
    rows: list[dict],
    key: str = "id",
    endpoint_key: str = "id",
    endpoint_labels: list[str] | None = None,
    src_field: str = "src",
    dst_field: str = "dst",
    replace: bool = False,
) -> dict

db.create_property_index("id")
db.upsert_nodes(["File"], [{"id": "f1"}, {"id": "f2"}])
db.upsert_edges("USES", [{"src": "f1", "dst": "f2", "id": "u1", "weight": 1}])
# {'created': 1, 'updated': 0, 'skipped': 0, 'skipped_rows': []}

DataFrame Integration¶

These methods convert between Grafeo and pandas/polars DataFrames. Requires pandas or polars to be installed (uv add pandas or uv add polars).

nodes_df()¶

Export all nodes as a pandas DataFrame. Columns: _id (int), _labels (list[str]), plus one column per unique property key. Missing properties are None. Lists, maps and durations are Python lists and dicts, with or without pyarrow installed (see Property types).

def nodes_df(self) -> pandas.DataFrame

df = db.nodes_df()
print(df[df["_labels"].apply(lambda l: "Person" in l)])

edges_df()¶

Export all edges as a pandas DataFrame. Columns: _id (int), _source (int), _target (int), _type (str), plus one column per unique property key. Missing properties are None.

def edges_df(self) -> pandas.DataFrame

df = db.edges_df()
print(df[df["_type"] == "KNOWS"])

import_df()¶

Bulk import nodes or edges from a pandas or polars DataFrame. Returns the number of rows imported.

def import_df(
    self,
    df: DataFrame,
    mode: str,                    # "nodes" or "edges"
    *,
    label: Optional[str | List[str]] = None,  # required for mode="nodes"
    edge_type: Optional[str] = None,          # required for mode="edges"
    source: str = "source",       # column name for source node IDs
    target: str = "target"        # column name for target node IDs
) -> int

Node import (mode='nodes'): each row becomes a node. The label parameter sets the label(s). All DataFrame columns become properties.

Edge import (mode='edges'): each row becomes an edge. The source and target columns must contain integer node IDs. Remaining columns become edge properties.

import pandas as pd

# Import nodes
people = pd.DataFrame({"name": ["Alix", "Gus"], "age": [30, 25]})
db.import_df(people, mode="nodes", label="Person")

# Import edges (source/target are node IDs)
edges = pd.DataFrame({"source": [0, 1], "target": [1, 0], "since": [2020, 2021]})
db.import_df(edges, mode="edges", edge_type="KNOWS")

Arrow Bulk Export¶

Zero-copy bulk export using Apache Arrow. These methods are faster than nodes_df()/edges_df() for large graphs because they build Arrow record batches directly in Rust, avoiding row-by-row Python object creation.

nodes_to_arrow(), edges_to_arrow(), nodes_to_pandas(), edges_to_pandas() and QueryResult.to_arrow() need pyarrow 14 or later. The arrow extra installs it, and a package that calls them can declare grafeo[arrow] as its dependency:

uv add "grafeo[arrow]"

Node schema: _id (uint64), _labels (list<utf8>), plus one column per unique property key.

Edge schema: _id (uint64), _type (utf8), _source (uint64), _target (uint64), plus one column per unique property key.

Property types¶

A property column's Arrow type follows its values, and so does result.to_arrow() for a query result:

A null is null at every level. Values of different types in one column, or in one list, are written as text (utf8), except integers and floats together, which are floats.

nodes_to_arrow()¶

Export all nodes as a pyarrow.Table. Requires pyarrow (uv add "grafeo[arrow]").

def nodes_to_arrow(self) -> pyarrow.Table

table = db.nodes_to_arrow()
print(table.schema)
print(table.to_pandas())

edges_to_arrow()¶

Export all edges as a pyarrow.Table. Requires pyarrow (uv add "grafeo[arrow]").

def edges_to_arrow(self) -> pyarrow.Table

table = db.edges_to_arrow()
print(table.filter(table.column("_type") == "KNOWS"))

nodes_to_polars()¶

Export all nodes as a polars.DataFrame. Requires polars (uv add polars), but does not require pyarrow. Uses the Arrow IPC format internally for zero-copy transfer.

def nodes_to_polars(self) -> polars.DataFrame

df = db.nodes_to_polars()
print(df.filter(pl.col("_labels").list.contains("Person")))

edges_to_polars()¶

Export all edges as a polars.DataFrame. Requires polars (uv add polars), but does not require pyarrow.

def edges_to_polars(self) -> polars.DataFrame

df = db.edges_to_polars()
print(df.filter(pl.col("_type") == "KNOWS"))

nodes_to_pandas()¶

Export all nodes as a pandas.DataFrame via the Arrow fast path. Requires both pandas and pyarrow (uv add pandas "grafeo[arrow]"). It keeps pyarrow's own conversion (a list is a numpy array); nodes_df() takes the same path when pyarrow is installed and returns lists and dicts.

def nodes_to_pandas(self) -> pandas.DataFrame

df = db.nodes_to_pandas()
print(df[df["_labels"].apply(lambda l: "Person" in l)])

edges_to_pandas()¶

Export all edges as a pandas.DataFrame via the Arrow fast path. Requires both pandas and pyarrow (uv add pandas "grafeo[arrow]").

def edges_to_pandas(self) -> pandas.DataFrame

df = db.edges_to_pandas()
print(df.groupby("_type").size())

Batch Operations¶

batch_create_nodes()¶

Bulk-insert nodes with a single vector property each. All nodes get the same label. Much faster than calling create_node() in a loop.

def batch_create_nodes(self, label: str, property: str, vectors: List[List[float]]) -> List[int]

Returns a list of created node IDs.

ids = db.batch_create_nodes("Doc", "embedding", [[1.0, 0.0], [0.0, 1.0]])

batch_create_nodes_with_props()¶

Batch-create nodes with full property maps. Each dict in the list is a complete set of properties for one node. Vector values are auto-inserted into matching vector indexes.

def batch_create_nodes_with_props(self, label: str | list[str], properties_list: List[Dict[str, Any]]) -> List[int]

Returns a list of created node IDs. The batch is one transaction: if a node breaks a constraint, none is created.

ids = db.batch_create_nodes_with_props("Person", [
    {"name": "Alix", "age": 30},
    {"name": "Gus", "age": 25},
])
db.batch_create_nodes_with_props(["Graph", "File"], [{"id": "f1"}, {"id": "f2"}])

batch_create_edges()¶

Batch-create edges, each with its own type and properties, from (src, dst, type) or (src, dst, type, properties) tuples. The batch is one transaction: if an edge names a node that does not exist or breaks the schema, none is created.

def batch_create_edges(self, edges: List[tuple]) -> List[int]

Returns a list of created edge IDs, in input order.

alix, gus = db.batch_create_nodes_with_props("Person", [{"name": "Alix"}, {"name": "Gus"}])
ids = db.batch_create_edges([
    (alix, gus, "KNOWS", {"since": 2020}),
    (gus, alix, "KNOWS"),
])

batch_vector_search()¶

Search for nearest neighbors of multiple query vectors in parallel across all available CPU cores.

def batch_vector_search(
    self,
    label: str,
    property: str,
    queries: List[List[float]],
    k: int,
    ef: Optional[int] = None,
    filters: Optional[Dict[str, Any]] = None
) -> List[List[Tuple[int, float]]]

Returns a list of results per query. Each result is a list of (node_id, distance) tuples.

results = db.batch_vector_search("Doc", "embedding", [[1.0, 0.0], [0.0, 1.0]], k=5)
for i, hits in enumerate(results):
    print(f"Query {i}: {hits}")

Search¶

vector_search()¶

Search for the k nearest neighbors of a query vector using the HNSW index.

def vector_search(
    self,
    label: str,
    property: str,
    query: List[float],
    k: int,
    ef: Optional[int] = None,
    filters: Optional[Dict[str, Any]] = None
) -> List[Tuple[int, float]]

Returns a list of (node_id, distance) tuples sorted by distance ascending (lower distance = more similar). The distance scale depends on the metric configured at index creation: cosine [0, 2], euclidean [0, inf), dot_product (negated, so lower = higher similarity), manhattan [0, inf).

results = db.vector_search("Doc", "embedding", [1.0, 0.0, 0.0], k=10, ef=200)
for node_id, distance in results:
    print(f"Node {node_id}: distance={distance:.4f}")

# With property filters
results = db.vector_search("Doc", "embedding", query, k=10, filters={"user_id": 42})

mmr_search()¶

Maximal Marginal Relevance search. Balances relevance to the query with diversity among results, avoiding redundant results in RAG pipelines.

def mmr_search(
    self,
    label: str,
    property: str,
    query: List[float],
    k: int,
    fetch_k: Optional[int] = None,       # initial candidates, default 4*k
    lambda_mult: Optional[float] = None,  # 0=diverse, 1=relevant, default 0.5
    ef: Optional[int] = None,
    filters: Optional[Dict[str, Any]] = None
) -> List[Tuple[int, float]]

Returns a list of (node_id, distance) tuples in MMR selection order. The distance values are identical to those returned by vector_search() for the same nodes (lower = more similar). The list ordering reflects MMR's relevance-diversity balance, not pure distance sorting.

results = db.mmr_search("Doc", "embedding", [1.0, 0.0, 0.0], k=4, lambda_mult=0.5)
for node_id, distance in results:
    print(f"Node {node_id}: distance={distance:.4f}")

text_search()¶

BM25 full-text search. Requires the text-index feature and a text index created with create_text_index().

def text_search(
    self,
    label: str,
    property: str,
    query: str,
    k: int,
    filters: Optional[Dict[str, Any]] = None
) -> List[Tuple[int, float]]

Returns a list of (node_id, score) tuples sorted by descending relevance (higher score = more relevant). BM25 scores are unbounded positive floats; compare them only within a single query's results.

filters takes the property filters of vector_search(): equality ({"city": "Berlin"}) and operators ({"rank": {"$gt": 19}}). Only matching nodes are searched, so up to k of them come back, scored as without the filters.

db.create_text_index("Article", "title")
results = db.text_search("Article", "title", "graph database", k=10)
for node_id, score in results:
    print(f"Node {node_id}: score={score:.4f}")

hybrid_search()¶

Combined text and vector search using Reciprocal Rank Fusion (RRF) or weighted fusion. Requires the hybrid-search feature. For best results, create both a text index (create_text_index()) and a vector index (create_vector_index()). If either index is missing, that source is silently omitted from fusion. With neither (no text index, and no vector index or no query_vector), it raises an error naming the indexes.

def hybrid_search(
    self,
    label: str,
    text_property: str,
    vector_property: str,
    query_text: str,
    k: int,
    query_vector: Optional[List[float]] = None,
    fusion: Optional[str] = None,          # "rrf" (default) or "weighted"
    weights: Optional[List[float]] = None, # [text_weight, vector_weight]
    rrf_k: Optional[int] = None,
    filters: Optional[Dict[str, Any]] = None
) -> List[Tuple[int, float]]

Returns a list of (node_id, score) tuples sorted by fused score descending (higher = more relevant). These are fusion scores, not distances. With RRF (default), scores are sum(1/(k+rank)) across sources. With weighted fusion, scores are normalized to [0, 1] and combined with explicit weights.

filters takes the property filters of vector_search(): equality ({"city": "Berlin"}) and operators ({"rank": {"$gt": 19}}). Both the text and the vector search keep only the matching nodes before fusion, so up to k matching nodes come back; text scores stay those of the whole index.

results = db.hybrid_search(
    "Article", "title", "embedding",
    "graph databases", k=10,
    query_vector=[1.0, 0.0, 0.0]
)

# Only articles from Berlin, from both searches
results = db.hybrid_search(
    "Article", "title", "embedding",
    "graph databases", k=10,
    query_vector=[1.0, 0.0, 0.0],
    filters={"city": "Berlin"}
)

Property Indexes¶

create_property_index()¶

Create an index on a node property for O(1) lookups.

def create_property_index(self, property: str) -> None

drop_property_index()¶

Remove a property index. Returns True if the index existed and was removed.

def drop_property_index(self, property: str) -> bool

has_property_index()¶

Check whether a property has an index.

def has_property_index(self, property: str) -> bool

find_nodes_by_property()¶

Find all nodes with a specific property value. O(1) if the property is indexed, O(n) otherwise.

def find_nodes_by_property(self, property: str, value: Any) -> List[int]

db.create_property_index("email")
node_ids = db.find_nodes_by_property("email", "alix@example.com")

Vector Index Management¶

create_vector_index()¶

Create an HNSW vector similarity index on a node property.

def create_vector_index(
    self,
    label: str,
    property: str,
    dimensions: Optional[int] = None,
    metric: Optional[str] = None,       # "cosine" (default), "euclidean", "dot_product", "manhattan"
    m: Optional[int] = None,            # HNSW links per node, default 16
    ef_construction: Optional[int] = None  # build beam width, default 128
) -> None

drop_vector_index()¶

Drop a vector index. Returns True if the index existed and was removed.

def drop_vector_index(self, label: str, property: str) -> bool

rebuild_vector_index()¶

Rebuild a vector index from scratch, preserving its configuration (dimensions, metric, M, ef_construction).

def rebuild_vector_index(self, label: str, property: str) -> None

Text Index Management¶

Requires the text-index feature.

create_text_index()¶

Create a BM25 text index on a node property. The index is automatically kept in sync as nodes are created, updated, or deleted. You do not need to call rebuild_text_index() after normal write operations.

def create_text_index(
    self,
    label: str,
    property: str,
    k1: Optional[float] = None,             # BM25 term frequency saturation, >= 0 (default 1.2)
    b: Optional[float] = None,              # BM25 length normalization, 0 to 1 (default 0.75)
    tokenizer: Optional[str] = None,        # "simple" (default), "standard" or "cjk_bigram"
    stop_words: Optional[List[str]] = None  # in place of the tokenizer's own
) -> None

The database keeps the options with the index: reopening it, recovering after a crash and rebuild_text_index() use them again. simple keeps words of at least 2 bytes without common English words, standard keeps every word (for Russian, Greek and other languages that separate words), and cjk_bigram also splits Chinese, Japanese and Korean text into overlapping pairs of characters. An unknown tokenizer, or k1 or b out of range, raises grafeo.GrafeoError (GRAFEO-V001).

db.create_text_index("Note", "body", tokenizer="cjk_bigram", k1=1.5, b=0.3)

drop_text_index()¶

Drop a text index. Returns True if the index existed and was removed.

def drop_text_index(self, label: str, property: str) -> bool

rebuild_text_index()¶

Rebuild a text index from scratch. Text indexes auto-sync on normal writes; you only need this after importing data through non-standard paths.

def rebuild_text_index(self, label: str, property: str) -> None

Transaction Methods¶

begin_transaction()¶

Start a new transaction. Returns a Transaction object that can be used as a context manager. The isolation_level parameter accepts a string: "read_committed" or "snapshot" (the default); "serializable" is refused for now: snapshot isolation is the strongest level until queries record their reads.

def begin_transaction(self, isolation_level: Optional[str] = None) -> Transaction

with db.begin_transaction() as tx:
    tx.execute("INSERT (:Person {name: 'Alix'})")
    tx.execute("INSERT (:Person {name: 'Gus'})")
    tx.commit()  # both nodes created atomically

# With explicit isolation level
with db.begin_transaction("read_committed") as tx:
    tx.execute("MATCH (n:Counter) SET n.val = n.val + 1")
    tx.commit()

begin_transaction_with_cdc()¶

Start a transaction with an explicit CDC (change data capture) override. When cdc_enabled is True, mutations in this transaction are tracked regardless of the database-level default. When False, tracking is disabled for this transaction only. Requires the cdc feature.

def begin_transaction_with_cdc(
    self,
    cdc_enabled: bool,
    isolation_level: Optional[str] = None
) -> Transaction

with db.begin_transaction_with_cdc(True) as tx:
    tx.execute("INSERT (:Person {name: 'Alix'})")
    tx.commit()
# This transaction's changes appear in node_history()

Schema Context¶

set_schema()¶

Set the current schema for subsequent execute() calls.

def set_schema(self, name: str) -> None

db.set_schema("reporting")
result = db.execute("SHOW GRAPH TYPES")  # only sees types in 'reporting'

reset_schema()¶

Clear the current schema context. Subsequent execute() calls will use the default namespace.

def reset_schema(self) -> None

current_schema()¶

Returns the current schema name, or None if no schema is set.

def current_schema(self) -> Optional[str]

RDF Loading¶

Load Turtle or N-Triples into the RDF store, from a file or from text, into the default graph or a named graph. Each call is one bulk write: the input is parsed first, nothing is inserted when a line does not parse (the error names the line), and the triples go to the WAL as one group, so a persistent database keeps all of them or none after a crash. A load reports no change data capture events. Each returns how many triples were new in its graph: a triple the graph holds already counts as nothing new, and the blank nodes of a document are new ones on each load. Requires the triple-store feature (in the default wheel).

A named graph is given by its IRI, without angle brackets. It is created when it does not exist, and holds the triples where INSERT DATA { GRAPH <iri> { ... } } puts them, so GRAPH <iri> { ... } in a query reads them.

load_rdf()¶

Load an RDF file. format is "turtle" or "ntriples" (also "ttl" and "nt"); by default the file's extension decides (.ttl, .nt).

def load_rdf(self, path: str | os.PathLike, format: str | None = None, graph: str | None = None) -> int

db.load_rdf("people.ttl")
db.load_rdf("cities.nt", graph="http://example.org/cities")

load_rdf_text()¶

Load RDF text in format ("turtle" or "ntriples").

def load_rdf_text(self, text: str, format: str, graph: str | None = None) -> int

turtle = """
@prefix ex: <http://example.org/> .
ex:alix ex:knows ex:gus ; ex:city "Amsterdam" .
"""
db.load_rdf_text(turtle, "turtle")  # 2
db.execute_sparql("SELECT ?who WHERE { <http://example.org/alix> <http://example.org/knows> ?who }")

SPARQL LOAD is not supported yet; use these calls instead.

SHACL Validation¶

Validate graph data against SHACL (Shapes Constraint Language) shapes. Requires the triple-store feature.

validate_shacl()¶

Validate the current graph against a SHACL shapes graph provided as a Turtle string. Returns a dict with the validation report.

def validate_shacl(self, shapes_graph: str) -> Dict[str, Any]

Return value keys:

Each result dict contains:

shapes = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix ex: <http://example.org/> .

ex:PersonShape a sh:NodeShape ;
    sh:targetClass ex:Person ;
    sh:property [
        sh:path ex:name ;
        sh:minCount 1 ;
        sh:datatype xsd:string ;
    ] .
"""

report = db.validate_shacl(shapes)
if report["conforms"]:
    print("All data is valid")
else:
    for r in report["results"]:
        print(f"{r['severity']}: {r['focus_node']} - {r['source_constraint_component']}")
    print(report["results_text"])

Admin Methods¶

The dicts these methods return gain keys only: a patch release can add a key, never remove or rename one, so read the keys you need rather than comparing whole dicts.

info()¶

Get database information. Returns a dict with keys: mode, node_count, edge_count, is_persistent, path, wal_enabled, version.

def info(self) -> Dict[str, Any]

detailed_stats()¶

Get detailed statistics. Returns a dict with keys: node_count, edge_count, label_count, edge_type_count, property_key_count, index_count, memory_bytes, disk_bytes. property_key_count counts the distinct property keys ever written, not the keys in use now: a key stays counted after every value of it is removed and after the transaction that wrote it rolled back.

def detailed_stats(self) -> Dict[str, Any]

memory_usage()¶

Returns a hierarchical memory usage breakdown. Walks all internal structures (store, indexes, MVCC chains, caches, string pools, buffer manager) and returns estimated heap bytes.

def memory_usage(self) -> Dict[str, Any]

Returns a dict with keys: total_bytes, store, indexes, mvcc, caches, string_pool, buffer_manager. Each sub-key is itself a dict with a detailed breakdown.

usage = db.memory_usage()
print(f"Total: {usage['total_bytes']} bytes")
print(f"Store: {usage['store']['total_bytes']} bytes")

schema()¶

Get schema information: the labels, edge types and property keys the data holds, each sorted by name. A label no node has any more, an edge type with no edge left and a property key nothing holds are left out. Returns a dict with keys:

• labels: dicts with name, count and properties
• edge_types: dicts with name, count, properties and endpoints
• property_keys: strings

Each property is a dict with its name and the types of its values (["STRING"], ["INT64", "FLOAT64"]). Each endpoint is a dict with a source label, a target label and the count of edges between nodes with them. It reads the whole graph, so call it once and keep the result rather than per query.

def schema(self) -> Dict[str, Any]

db.execute("INSERT (:Person {name: 'Alix'})-[:LIVES_IN {since: 2020}]->(:City {name: 'Amsterdam'})")
schema = db.schema()
schema["labels"][1]
# {'name': 'Person', 'count': 1, 'properties': [{'name': 'name', 'types': ['STRING']}]}
schema["edge_types"][0]["endpoints"]
# [{'source': 'Person', 'target': 'City', 'count': 1}]

validate()¶

Validate database integrity. Returns a list of error dicts (empty list means valid). Each error dict has keys: code, message, context.

def validate(self) -> List[Dict[str, str]]

errors = db.validate()
if not errors:
    print("Database is valid")
else:
    for err in errors:
        print(f"[{err['code']}] {err['message']}")

wal_status()¶

Returns WAL (Write-Ahead Log) status. Returns a dict with keys: enabled, path, size_bytes, record_count, last_checkpoint, current_epoch.

def wal_status(self) -> Dict[str, Any]

wal = db.wal_status()
print(f"WAL size: {wal['size_bytes']} bytes")

set_checkpoint_triggers()¶

Let the database checkpoint by itself: every interval_secs seconds, and once the WAL holds wal_bytes bytes since the last checkpoint. Call it without arguments to turn automatic checkpoints off (the default). For this open only; an in-memory or read-only database raises.

def set_checkpoint_triggers(self, interval_secs: Optional[float] = None, wal_bytes: Optional[int] = None) -> None

db.set_checkpoint_triggers(interval_secs=300, wal_bytes=64 * 1024 * 1024)

checkpoint_status()¶

How the automatic checkpoints fare. Returns a dict with keys: interval_secs and wal_bytes (the triggers, None when off), last_checkpoint (Unix seconds of the last checkpoint of any kind), failures (automatic checkpoints that failed since the last that succeeded), last_error, last_failure and stopped (no automatic checkpoint can succeed until a reopen). A failed automatic checkpoint is also logged and retried after a pause.

def checkpoint_status(self) -> Dict[str, Any]

save()¶

Save the database to a file path. For in-memory databases, creates a new persistent copy. For file-backed databases, creates a copy at the new path. The copy is a single file, whatever the extension of the path, and save() fails if the path already exists. The original database remains unchanged.

def save(self, path: str) -> None

db = GrafeoDB()  # in-memory
db.create_node(["Person"], {"name": "Alix"})
db.save("./mydb")  # persist to disk

to_memory()¶

Create an independent in-memory copy of this database. Changes to the copy do not affect the original. The copy has every graph with its data, the schema and constraints, and the property, vector and text indexes: what reopening the database from a checkpoint would give.

def to_memory(self) -> GrafeoDB

file_db = GrafeoDB("./production.grafeo")
test_db = file_db.to_memory()  # safe copy for experiments, indexes included

compact()¶

Compacts the database: writes a checkpoint of a persistent database (an in-memory or read-only one writes none), drops the old versions no open transaction can see any more, compresses the property values held decoded in memory, and returns what it did as a dict: checkpointed, versions_collected, rows_compressed and duration_ms. Since 0.6.0 the database keeps one store: compact() no longer builds a separate columnar one (see Compact Store). Raises if the checkpoint fails, or after close().

def compact(self) -> dict

db = grafeo.GrafeoDB()
db.execute("INSERT (:Person {name: 'Alix', age: 30})")
db.execute("INSERT (:Person {name: 'Gus', age: 25})")

report = db.compact()  # {'checkpointed': False, 'versions_collected': 0, 'rows_compressed': 2, 'duration_ms': 0}

result = db.execute("MATCH (p:Person) RETURN p.name")
db.execute("INSERT (:Person {name: 'Vincent'})")

close()¶

Close the database, flushing any pending writes.

def close(self) -> None

clear_plan_cache()¶

Clear all cached query plans, forcing re-parsing and re-optimization on next execution. Called automatically after DDL operations, but can be invoked manually after external schema changes.

def clear_plan_cache(self) -> None

Properties¶

Temporal Queries¶

get_node_at_epoch()¶

Get a node as it existed at a specific historical epoch. Returns None if the node did not exist at that epoch, and raises GrafeoError for an epoch older than the history the database keeps.

def get_node_at_epoch(self, id: int, epoch: int) -> Optional[Node]

get_edge_at_epoch()¶

Get an edge as it existed at a specific historical epoch. Returns None if the edge did not exist at that epoch, and raises GrafeoError for an epoch older than the history the database keeps.

def get_edge_at_epoch(self, id: int, epoch: int) -> Optional[Edge]

get_node_history()¶

Get the version history of a node. Returns a list of (created_epoch, deleted_epoch, node) tuples.

def get_node_history(self, id: int) -> List[Tuple[int, Optional[int], Node]]

get_edge_history()¶

Get the version history of an edge. Returns a list of (created_epoch, deleted_epoch, edge) tuples.

def get_edge_history(self, id: int) -> List[Tuple[int, Optional[int], Edge]]

get_node_property_at_epoch()¶

Returns a property value as it existed at a specific epoch, or None if the property did not exist then. Raises GrafeoError for an epoch older than the history the database keeps.

def get_node_property_at_epoch(self, id: int, key: str, epoch: int) -> Optional[Any]

get_node_property_history()¶

Returns the full version timeline for a single property: list of (epoch, value) tuples.

def get_node_property_history(self, id: int, key: str) -> List[Tuple[int, Any]]

get_all_node_property_history()¶

Returns version history for all properties: dict mapping property names to lists of (epoch, value) tuples, within the history the database keeps.

def get_all_node_property_history(self, id: int) -> Dict[str, List[Tuple[int, Any]]]

current_epoch()¶

Returns the current epoch of the database. The epoch increments with each committed transaction.

def current_epoch(self) -> int

Algorithms¶

Access graph algorithms via the algorithms property (not a method call). All algorithms run directly on the Rust graph store with no data copying. Requires the algos feature.

algos = db.algorithms  # property, not db.algorithms()

Traversal¶

Shortest Paths¶

dijkstra()¶

Dijkstra's algorithm. When target is provided, returns (distance, path) or None. When target is omitted, returns a dict mapping node IDs to distances.

def dijkstra(
    self,
    source: int,
    target: Optional[int] = None,
    weight: Optional[str] = None
) -> Union[Dict[int, float], Tuple[float, List[int]], None]

# Single target
result = db.algorithms.dijkstra(1, 5, weight="cost")
if result:
    distance, path = result
    print(f"Distance: {distance}, Path: {path}")

# All distances from source
distances = db.algorithms.dijkstra(1)

floyd_warshall()¶

All-pairs shortest paths. Returns a dict mapping (source, target) tuples to distances.

def floyd_warshall(self, weight: Optional[str] = None) -> Dict[Tuple[int, int], float]

Centrality¶

Community Detection¶

Clustering and Triangles¶

Structure Analysis¶

Minimum Spanning Tree¶

Network Flow¶

QueryResult¶

Returned by execute() and other query methods. Iterable: each row is a dict keyed by column name.

Result Properties¶

Result Methods¶

result = db.execute("MATCH (p:Person) RETURN p.name, p.age")

# Iterate rows
for row in result:
    print(row["p.name"])

# Convert to DataFrame
df = result.to_pandas()

# Access metrics
if result.execution_time_ms:
    print(f"Query took {result.execution_time_ms:.2f}ms")

Example¶

import grafeo

db = grafeo.GrafeoDB()

# Execute queries
db.execute("INSERT (:Person {name: 'Alix', age: 30})")

result = db.execute("MATCH (p:Person) RETURN p.name")
for row in result:
    print(row['p.name'])

# Use transactions
with db.begin_transaction() as tx:
    tx.execute("INSERT (:Person {name: 'Gus'})")
    tx.commit()

# DataFrame integration
df = db.nodes_df()
print(df.head())

# Graph algorithms
pr = db.algorithms.pagerank()
path = db.algorithms.dijkstra(1, 5)