API Reference

Graph Models and Database

class gestaltdb.graphdb.Edge(edge_id=None, source=None, target=None, properties=None)

Bases: object

Directed graph edge with source, target, and properties.

Parameters:

• edge_id – Optional stable edge identifier. A UUID is generated when omitted.

• source – Source node ID or Node instance.

• target – Target node ID or Node instance.

• properties – Optional edge attributes. Typed traversal reads properties["type"].

Examples

>>> Edge(edge_id="d1-p1", source="drug-1", target="protein-1").source
'drug-1'

__init__(edge_id=None, source=None, target=None, properties=None)

If no edge_id is provided, generate a UUID.

classmethod from_dict(data)

Factory from dictionary.

Parameters:

data (dict)

property get_id

Unique identifier for this edge.

property get_id_bytes

Return the edge ID encoded as UTF-8 bytes.

Examples

>>> Edge(edge_id="d1-p1").get_id_bytes
b'd1-p1'

property get_type

Return the typed traversal edge type.

Examples

>>> Edge(properties={"type": "drug-to-protein"}).get_type
'drug-to-protein'

to_dict()

Convert to a dictionary for serialization.

class gestaltdb.graphdb.GraphDB(store, serializer, indexed_node_properties=None, indexed_edge_properties=None)

Bases: object

High-level interface to manage Node/Edge storing, retrieval, and indexing.

Parameters:

• store (KVStore)

• serializer (Serializer)

• indexed_node_properties (Optional[list[str]])

• indexed_edge_properties (Optional[list[str]])

__init__(store, serializer, indexed_node_properties=None, indexed_edge_properties=None)

Initialize a graph database wrapper.

Parameters:

• store (KVStore) – KVStore instance such as LMDBStore, LevelDBStore, or PyRexStore.

• serializer (Serializer) – Serializer for node, edge, and adjacency payloads.

• indexed_node_properties (Optional[list[str]]) – Optional exact-match node property indexes to maintain for future writes.

• indexed_edge_properties (Optional[list[str]]) – Optional exact-match edge property indexes to maintain for future writes.

Examples

>>> from gestaltdb.kvstores import LMDBStore
>>> from gestaltdb.serializers import PickleSerializer
>>> graph = GraphDB(LMDBStore(path="/tmp/example"), PickleSerializer(), indexed_node_properties=["name"])

bfs(start_node_id, direction='any', edge_key_serializer=<function GraphDB.<lambda>>, node_key_serializer=<function GraphDB.<lambda>>)

Returns a list of node_ids in BFS order starting from start_node_id. Demonstrates how adjacency is used for graph traversal.

Parameters:

start_node_id (bytes)

Return type:

list[str]

build_sampler_snapshot(output_path, *, source_db_reference=True, source_db_path=None, source_db_path_mode='relative', **kwargs)

Build a read-optimized array sampler snapshot from this graph.

Parameters:

• output_path – Directory where snapshot arrays and metadata are written.

• **kwargs – Options forwarded to SamplerSnapshot.build.

• source_db_reference (bool)

• source_db_path_mode (str)

Returns:

A loaded SamplerSnapshot instance.

close()

Close the underlying key-value store.

Examples

>>> graph_db.close()

count_edges_by_property(property_name, value)

Return the number of edges currently indexed for an exact property value.

Parameters:

property_name (str)

Return type:

int

count_edges_by_property_range(property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return the number of edges indexed in a scalar property range.

Parameters:

• property_name (str)

• include_start (bool)

• include_end (bool)

Return type:

int

count_edges_by_type(edge_type)

Return the number of edges currently indexed for a relationship type.

Parameters:

edge_type (str)

Return type:

int

count_edges_by_type_property(edge_type, property_name, value)

Return the number of edges indexed for a type and exact property value.

Parameters:

• edge_type (str)

• property_name (str)

Return type:

int

count_edges_by_type_property_range(edge_type, property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return the number of edges indexed for a type/property range.

Parameters:

• edge_type (str)

• property_name (str)

• include_start (bool)

• include_end (bool)

Return type:

int

count_nodes_by_label(label)

Return the number of nodes currently indexed for a label.

Parameters:

label (str)

Return type:

int

count_nodes_by_label_property(label, property_name, value)

Return the number of nodes indexed for a label and exact property value.

Parameters:

• label (str)

• property_name (str)

Return type:

int

count_nodes_by_label_property_range(label, property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return the number of nodes indexed for a label/property range.

Parameters:

• label (str)

• property_name (str)

• include_start (bool)

• include_end (bool)

Return type:

int

count_nodes_by_property(property_name, value)

Return the number of nodes currently indexed for an exact property value.

Parameters:

property_name (str)

Return type:

int

count_nodes_by_property_range(property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return the number of nodes indexed in a scalar property range.

Parameters:

• property_name (str)

• include_start (bool)

• include_end (bool)

Return type:

int

classmethod create(path, *, backend='pyrex', serializer='json', backend_options=None, indexed_node_properties=None, indexed_edge_properties=None, overwrite=False)

Create a self-describing graph store and return an open handle.

Return type:

GraphDB

create_edge_property_index(property_name)

Register and rebuild an exact-match edge property index.

Parameters:

property_name (str) – Edge property to index for exact-match lookup.

Returns:

Number of existing edges added to the index.

Examples

>>> graph_db.create_edge_property_index("score")
7

create_node_property_index(property_name)

Register and rebuild an exact-match node property index.

Parameters:

property_name (str) – Node property to index for exact-match lookup.

Returns:

Number of existing nodes added to the index.

Examples

>>> graph_db.create_node_property_index("kind")
10

delete_edge(edge_id, edge_key_serializer=<function GraphDB.<lambda>>)

Removes the edge from the edge store, and from adjacency of both source and target nodes. If either node doesn’t exist, we skip gracefully.

Parameters:

edge_id (str)

delete_node(node_id)

Delete a node by byte key.

Parameters:

node_id – Node ID bytes.

Examples

>>> graph_db.delete_node(b"drug-1")

edge_key_to_bytes(edge_key)

Normalize an edge key to bytes.

Parameters:

edge_key – String or bytes edge key.

Returns:

UTF-8 encoded bytes.

Examples

>>> GraphDB.edge_key_to_bytes(None, "d1-p1")
b'd1-p1'

edge_type(edge)

Return the type used by typed traversal for an edge.

Parameters:

edge (Edge) – Edge to inspect.

Returns:

Edge type string, or None.

Examples

>>> GraphDB.edge_type(None, Edge(properties={"type": "drug-to-protein"}))
'drug-to-protein'

edges_by_edge_type(node_id, edge_type, direction='out')

Return edge IDs connected by a specific edge type.

Parameters:

• node_id – Node ID as string or bytes.

• edge_type (str) – Edge type to traverse.

• direction (str) – "out", "in", or "any".

Returns:

List of edge ID bytes.

Examples

>>> graph_db.edges_by_edge_type("drug-1", "drug-to-protein")

edges_by_property(property_name, value)

Return edges using an exact-match property index.

Parameters:

• property_name (str) – Indexed edge property name.

• value – Exact property value to match.

Returns:

List of decoded Edge objects.

Examples

>>> graph_db.edges_by_property("score", 1)

edges_by_property_range(property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return edges using a scalar property range index.

Parameters:

• property_name (str)

• include_start (bool)

• include_end (bool)

edges_by_type(edge_type)

Return edges using the relationship type catalog.

Parameters:

edge_type (str) – Relationship type stored in edge.properties["type"].

Returns:

List of decoded Edge objects.

Examples

>>> graph_db.edges_by_type("drug-to-protein")

edges_by_type_property(edge_type, property_name, value)

Return edges using the composite type/property exact-match index.

Parameters:

• edge_type (str)

• property_name (str)

edges_by_type_property_range(edge_type, property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return edges using a composite type/property range index.

Parameters:

• edge_type (str)

• property_name (str)

• include_start (bool)

• include_end (bool)

get_adjacency_list(node_id, direction='forward', return_raw=False)

Returns the list of edge IDs connected to node_id. If none found, returns an empty list.

Parameters:

• node_id (bytes) – a string representing the node_id

• direction – ‘forward’, ‘backward’ or ‘any’ -> controls whether the source, target, or un-directed adjacency of the node will be returned.

• return_raw – if this flag is true it will return the data as they are stored (e.g., a dictionary of ‘source’ and ‘target’ lists. )

Return type:

list[str]

get_edge(edge_id)

Return an edge by byte key.

Parameters:

edge_id – Edge ID bytes as stored in the backend.

Returns:

The decoded edge, or None when absent.

Return type:

Edge

Examples

>>> graph_db.get_edge(b"d1-p1")

get_node(node_id)

Return a node by byte key.

Parameters:

node_id – Node ID bytes as stored in the backend.

Returns:

The decoded node, or None when absent.

Return type:

Node

Examples

>>> graph_db.get_node(b"drug-1")

get_node_keys_generator(num_nodes=None, key_offset=None)

Yield node keys from the backing store.

Parameters:

• num_nodes – Optional maximum number of keys to yield.

• key_offset – Optional starting key.

Returns:

Generator of node key bytes.

Examples

>>> list(graph_db.get_node_keys_generator(num_nodes=10))

get_nodes(node_ids)

Use store.get_nodes_bulk(…) and deserialize each one. Return a list of Node (in the same order as node_ids, or possibly just all found).

Parameters:

node_ids (list[str])

Return type:

list[Node]

get_typed_adjacency(node_id, edge_type, direction='out')

Return typed adjacency records with clean direction semantics.

out means source -> target, in means target -> source, and any returns the union of both directions.

Parameters:

• edge_type (str)

• direction (str)

index_statistics()

Return persisted index definitions visible to the query planner.

Return type:

dict[str, object]

ingest_arrow(node_ids, edge_ids, sources, targets, edge_types, *, ingestion_mode=ColumnarIngestionMode.ENTITY_COLUMNS, index_mode=IndexMaintenanceMode.DEFER_REBUILD, node_values=None, edge_values=None, labels=None, node_properties=None, edge_properties=None, native=True, chunk_size=100000, progress=False)

Ingest a graph from Arrow-like columns with bulk-load defaults.

ENTITY_COLUMNS builds payloads from structured node/edge columns. SERIALIZED_PAYLOADS expects node_values and edge_values to contain serializer-compatible payload bytes.

Example

>>> graph.ingest_arrow(node_ids, edge_ids, sources, targets, edge_types, node_properties={"kind": kinds})

Parameters:

• ingestion_mode (str | ColumnarIngestionMode)

• index_mode (str | IndexMaintenanceMode)

• node_properties (dict[str, object] | None)

• edge_properties (dict[str, object] | None)

• native (bool)

• chunk_size (int)

• progress (bool)

Return type:

dict[str, object]

ingest_edges_arrow(edge_ids, sources, targets, edge_types, edge_values, *, append_only=True, native=True, chunk_size=100000, index_mode='maintain', progress=False)

Ingest typed edges from Arrow-like columns.

edge_values is required and must contain serialized edge payloads compatible with the current GraphDB serializer. This ingestion path writes edge records and typed adjacency records only; it intentionally skips legacy adjacency blobs for append-friendly bulk loading.

Parameters:

• edge_ids – Arrow-like or Python column of edge IDs.

• sources – Arrow-like or Python column of source node IDs.

• targets – Arrow-like or Python column of target node IDs.

• edge_types – Arrow-like or Python column of typed traversal labels.

• edge_values – Arrow-like or Python column of serialized edge bytes.

• append_only (bool) – Columnar ingestion currently requires True.

• native (bool) – Use native backend columnar ingestion when available.

• chunk_size (int) – Maximum rows per backend write.

• index_mode (str) – "maintain" updates secondary indexes immediately; "defer" writes edge records and typed adjacency and marks secondary indexes stale.

• progress (bool)

Returns:

Number of ingested edges.

ingest_edges_arrow_entities(edge_ids, sources, targets, edge_types, *, properties=None, append_only=True, native=True, chunk_size=100000, index_mode='maintain', progress=False)

Ingest edge entity columns from Arrow-like columns.

Parameters:

• properties (dict[str, object] | None)

• append_only (bool)

• native (bool)

• chunk_size (int)

• index_mode (str)

• progress (bool)

ingest_edges_polars(df, *, edge_id='edge_id', source='source', target='target', edge_type='edge_type', edge_value='edge_value', append_only=True, native=True, chunk_size=100000, index_mode='maintain', progress=False)

Ingest typed edges from a Polars DataFrame.

The edge_value column is required and must contain serialized edge payload bytes compatible with the current GraphDB serializer.

Parameters:

• edge_id (str)

• source (str)

• target (str)

• edge_type (str)

• edge_value (str)

• append_only (bool)

• native (bool)

• chunk_size (int)

• index_mode (str)

• progress (bool)

ingest_edges_polars_entities(df, *, edge_id='edge_id', source='source', target='target', edge_type='edge_type', property_columns=None, append_only=True, native=True, chunk_size=100000, index_mode='maintain', progress=False)

Ingest edge entity columns from a Polars DataFrame.

With JSONSerializer, edge payloads are built with Polars struct JSON encoding. Other serializers fall back to Python Edge serialization.

Parameters:

• edge_id (str)

• source (str)

• target (str)

• edge_type (str)

• property_columns (list[str] | tuple[str, ...] | None)

• append_only (bool)

• native (bool)

• chunk_size (int)

• index_mode (str)

• progress (bool)

ingest_nodes_arrow(node_ids, node_values, *, native=True, chunk_size=100000, append_only=False, index_mode='maintain', progress=False)

Ingest attributed nodes from Arrow-like columns.

node_values is required and must contain serialized node payloads compatible with the current GraphDB serializer.

Parameters:

• node_ids – Arrow-like or Python column of node IDs.

• node_values – Arrow-like or Python column of serialized node bytes.

• native (bool) – Use native backend columnar ingestion when available.

• chunk_size (int) – Maximum rows per backend write.

• append_only (bool) – Skip existing-node index deletion for known-new nodes.

• index_mode (str) – "maintain" updates secondary indexes immediately; "defer" writes node records and marks indexes stale.

• progress (bool)

Returns:

Number of ingested nodes.

ingest_nodes_arrow_entities(node_ids, *, labels=None, properties=None, native=True, chunk_size=100000, append_only=False, index_mode='maintain', progress=False)

Ingest node entity columns from Arrow-like columns.

Parameters:

• properties (dict[str, object] | None)

• native (bool)

• chunk_size (int)

• append_only (bool)

• index_mode (str)

• progress (bool)

ingest_nodes_polars(df, *, node_id='node_id', node_value='node_value', native=True, chunk_size=100000, append_only=False, index_mode='maintain', progress=False)

Ingest attributed nodes from a Polars DataFrame.

The node_value column is required and must contain serialized node payload bytes compatible with the current GraphDB serializer.

Parameters:

• node_id (str)

• node_value (str)

• native (bool)

• chunk_size (int)

• append_only (bool)

• index_mode (str)

• progress (bool)

ingest_nodes_polars_entities(df, *, node_id='node_id', labels='labels', property_columns=None, native=True, chunk_size=100000, append_only=False, index_mode='maintain', progress=False)

Ingest node entity columns from a Polars DataFrame.

With JSONSerializer, node payloads are built with Polars struct JSON encoding. Other serializers fall back to Python Node serialization.

Parameters:

• node_id (str)

• labels (str | list[str] | tuple[str, ...] | None)

• property_columns (list[str] | tuple[str, ...] | None)

• native (bool)

• chunk_size (int)

• append_only (bool)

• index_mode (str)

• progress (bool)

ingest_polars(node_df, edge_df, *, ingestion_mode=ColumnarIngestionMode.ENTITY_COLUMNS, index_mode=IndexMaintenanceMode.DEFER_REBUILD, node_id='node_id', node_value='node_value', labels='labels', node_property_columns=None, edge_id='edge_id', source='source', target='target', edge_type='edge_type', edge_value='edge_value', edge_property_columns=None, native=True, chunk_size=100000, progress=False)

Ingest a graph from Polars DataFrames with bulk-load defaults.

By default this uses structured entity columns, defers secondary index maintenance during ingestion, and immediately performs a one-pass rebuild before returning. Typed adjacency is maintained during ingestion, so traversal remains correct throughout.

Example

>>> graph.ingest_polars(nodes, edges, node_property_columns=["kind"], edge_property_columns=["score"])

Parameters:

• ingestion_mode (str | ColumnarIngestionMode)

• index_mode (str | IndexMaintenanceMode)

• node_id (str)

• node_value (str)

• labels (str | list[str] | tuple[str, ...] | None)

• node_property_columns (list[str] | tuple[str, ...] | None)

• edge_id (str)

• source (str)

• target (str)

• edge_type (str)

• edge_value (str)

• edge_property_columns (list[str] | tuple[str, ...] | None)

• native (bool)

• chunk_size (int)

• progress (bool)

Return type:

dict[str, object]

iter_edge_ids_by_property(property_name, value)

Yield edge IDs from an exact-match edge property index.

Parameters:

• property_name (str) – Indexed edge property name.

• value – Exact property value to match.

Yields:

Edge ID bytes matching the property value.

Examples

>>> list(graph_db.iter_edge_ids_by_property("score", 1))
[b'e1']

iter_edge_ids_by_property_range(property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Yield edge IDs from a scalar property range index.

Parameters:

• property_name (str)

• include_start (bool)

• include_end (bool)

iter_edge_ids_by_type(edge_type)

Yield edge IDs from the relationship type catalog.

Parameters:

edge_type (str) – Relationship type stored in edge.properties["type"].

Yields:

Edge ID bytes with the requested relationship type.

Examples

>>> list(graph_db.iter_edge_ids_by_type("drug-to-protein"))
[b'd1-p1']

iter_edge_ids_by_type_property(edge_type, property_name, value)

Yield edge IDs from the composite type/property exact-match index.

Parameters:

• edge_type (str)

• property_name (str)

iter_edge_ids_by_type_property_range(edge_type, property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Yield edge IDs from a composite type/property range index.

Parameters:

• edge_type (str)

• property_name (str)

• include_start (bool)

• include_end (bool)

iter_node_ids_by_label(label)

Yield node IDs from the label index.

Parameters:

label (str) – Node label to scan.

Yields:

Node ID bytes with the requested label.

Examples

>>> list(graph_db.iter_node_ids_by_label("Drug"))
[b'drug-1']

iter_node_ids_by_label_property(label, property_name, value)

Yield node IDs from the composite label/property exact-match index.

Parameters:

• label (str)

• property_name (str)

iter_node_ids_by_label_property_range(label, property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Yield node IDs from a composite label/property range index.

Parameters:

• label (str)

• property_name (str)

• include_start (bool)

• include_end (bool)

iter_node_ids_by_property(property_name, value)

Yield node IDs from an exact-match property index.

Parameters:

• property_name (str) – Indexed node property name.

• value – Exact property value to match.

Yields:

Node ID bytes matching the property value.

Examples

>>> list(graph_db.iter_node_ids_by_property("kind", "drug"))
[b'drug-1']

iter_node_ids_by_property_range(property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Yield node IDs from a scalar property range index.

Parameters:

• property_name (str)

• include_start (bool)

• include_end (bool)

iter_typed_adjacency(node_id, edge_type, direction='out')

Yield typed adjacency records with clean direction semantics.

Parameters:

• node_id – Node ID as string or bytes.

• edge_type (str) – Edge type to traverse.

• direction (str) – "out", "in", or "any".

Yields:

Typed adjacency records containing edge, neighbor, source, target, edge type, and concrete direction fields.

Examples

>>> graph_db.iter_typed_adjacency("drug-1", "drug-to-protein")

key_to_string(key)

Normalize a key to a string.

Parameters:

key – String or UTF-8 bytes key.

Returns:

String key.

Examples

>>> GraphDB.key_to_string(None, b"drug-1")
'drug-1'

property manifest: dict

Return the loaded or generated manifest for this graph.

neighbors_by_edge_type(node_id, edge_type, direction='out')

Return neighbor IDs connected by a specific edge type.

Parameters:

• node_id – Node ID as string or bytes.

• edge_type (str) – Edge type to traverse.

• direction (str) – "out", "in", or "any".

Returns:

List of neighbor ID bytes.

Examples

>>> graph_db.neighbors_by_edge_type("drug-1", "drug-to-protein")

node_key_to_bytes(node_key)

Normalize a node key to bytes.

Parameters:

node_key – String or bytes node key.

Returns:

UTF-8 encoded bytes.

Examples

>>> GraphDB.node_key_to_bytes(None, "drug-1")
b'drug-1'

nodes_by_label(label)

Return nodes with a label using the label index.

Parameters:

label (str) – Node label to scan.

Returns:

List of decoded Node objects.

Examples

>>> graph_db.nodes_by_label("Drug")

nodes_by_label_property(label, property_name, value)

Return nodes using the composite label/property exact-match index.

Parameters:

• label (str)

• property_name (str)

nodes_by_label_property_range(label, property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return nodes using a composite label/property range index.

Parameters:

• label (str)

• property_name (str)

• include_start (bool)

• include_end (bool)

nodes_by_property(property_name, value)

Return nodes using an exact-match property index.

Parameters:

• property_name (str) – Indexed node property name.

• value – Exact property value to match.

Returns:

List of decoded Node objects.

Examples

>>> graph_db.nodes_by_property("kind", "drug")

nodes_by_property_range(property_name, start_value=None, end_value=None, include_start=True, include_end=True)

Return nodes using a scalar property range index.

Parameters:

• property_name (str)

• include_start (bool)

• include_end (bool)

classmethod open(path, *, backend_options=None, validate_manifest=True)

Open a self-describing graph store from a directory.

Return type:

GraphDB

put_adjacency_list(node_id, edges_list)

Stores the adjacency list for node_id.

Parameters:

• node_id (str)

• edges_list (list[str])

put_edge(edge, update_adjacency=True)

Store an edge and update adjacency indexes.

Parameters:

• edge (Edge) – Edge to serialize and write.

• update_adjacency – Whether to update the legacy untyped adjacency list.

Examples

>>> graph_db.put_edge(Edge(source="drug-1", target="protein-1"))

put_edges_bulk(edges, check_existing=True)

Store multiple edges and update adjacency indexes in bulk.

Parameters:

• edges (List[Edge]) – Edges to write.

• check_existing (bool) – When true, read existing edges first and remove stale typed adjacency and sorted index records before replacement. Set to false for append-only ingestion with new edge IDs.

Examples

>>> graph_db.put_edges_bulk([Edge(source="drug-1", target="protein-1")], check_existing=False)

put_node(node)

Store a node.

Parameters:

node (Node) – Node to serialize and write.

Examples

>>> graph_db.put_node(Node(node_id="drug-1"))

put_nodes(nodes)

Store multiple nodes and maintain label/property indexes.

Parameters:

nodes (list[Node]) – Nodes to serialize and write.

Examples

>>> graph_db.put_nodes([Node(node_id="drug-1", labels=["Drug"])])

query(cypher, parameters=None)

Execute a supported read-only Cypher query.

Parameters:

• cypher (str) – Query text in the supported GestaltDB Cypher subset.

• parameters (dict[str, object] | None) – Optional Cypher parameter values keyed without the leading $.

Returns:

gestaltdb.cypher.QueryResult containing projected records.

Examples

>>> graph_db.query('MATCH (n:Drug) RETURN n')
>>> graph_db.query('MATCH (a {id: "drug-1"})-[:drug-to-protein]->(b) RETURN a, b')

range_query_nodes(property_name, start_val, end_val)

Example stub: You might rely on the underlying store to handle indexing for nodes.

Parameters:

property_name (str)

rebuild_deferred_indexes()

Rebuild secondary indexes marked stale by deferred bulk ingestion.

Returns:

Mapping from rebuilt index family/property to entries written.

Return type:

dict[str, int]

rebuild_edge_indexes(*, edge_types=True, properties=None, batch_size=100000)

Rebuild requested edge secondary indexes in one edge scan.

Parameters:

• edge_types (bool)

• properties (list[str] | tuple[str, ...] | set[str] | None)

• batch_size (int)

Return type:

dict[str, int]

rebuild_edge_property_index(property_name)

Rebuild an exact-match edge property index from stored edges.

Parameters:

property_name (str) – Edge property to index.

Returns:

Number of indexed edge records.

Examples

>>> graph_db.rebuild_edge_property_index("score")
7

rebuild_label_index()

Rebuild the node label index from stored nodes.

Returns:

Number of label index entries written.

Examples

>>> graph_db.rebuild_label_index()
12

rebuild_node_indexes(*, labels=True, properties=None, batch_size=100000)

Rebuild requested node secondary indexes in one node scan.

Parameters:

• labels (bool)

• properties (list[str] | tuple[str, ...] | set[str] | None)

• batch_size (int)

Return type:

dict[str, int]

rebuild_node_property_index(property_name)

Rebuild an exact-match node property index from stored nodes.

Parameters:

property_name (str) – Node property to index.

Returns:

Number of indexed node records.

Examples

>>> graph_db.rebuild_node_property_index("name")
3

rebuild_relationship_type_index()

Rebuild the relationship type catalog from stored edges.

Returns:

Number of typed edge records indexed.

Examples

>>> graph_db.rebuild_relationship_type_index()
20

rebuild_typed_adjacency()

Rebuild typed adjacency indexes from stored edge records.

Returns:

Number of typed edges indexed.

Examples

>>> graph_db.rebuild_typed_adjacency()

sample_neighbors(node_id, edge_type, direction='out', sample_size=10, rng=None)

Sample typed neighbors using reservoir sampling.

Parameters:

• node_id – Node ID as string or bytes.

• edge_type (str) – Edge type to traverse.

• direction (str) – "out", "in", or "any".

• sample_size (int) – Maximum number of records to return.

• rng – Optional random number generator with randrange.

Returns:

List of typed adjacency records.

Examples

>>> graph_db.sample_neighbors("drug-1", "drug-to-protein", sample_size=2)

sample_typed_paths(seed_ids, pattern, rng=None)

Sample paths that follow an ordered typed edge pattern.

Parameters:

• seed_ids – Starting node IDs as strings or bytes.

• pattern (SamplingPattern | list[dict]) – SamplingPattern or list of dictionaries such as {"edge_type": "drug-to-protein", "direction": "out", "sample_size": 2}.

• rng – Optional random number generator with randrange.

Returns:

List of dictionaries with seed and sampled path records.

Examples

>>> from gestaltdb.sampling import SamplingHop, SamplingPattern
>>> pattern = SamplingPattern([SamplingHop("drug-to-protein", sample_size=2)])
>>> graph_db.sample_typed_paths(["drug-1"], pattern)

sample_typed_subgraph(seed_ids, pattern, rng=None)

Sample and materialize a typed subgraph around seed nodes.

Parameters:

• seed_ids – Starting node IDs as strings or bytes.

• pattern (SamplingPattern | list[dict]) – SamplingPattern or list of dictionary hop configs.

• rng – Optional random number generator with randrange.

Returns:

Dictionary with nodes, edges, and paths entries.

Examples

>>> pattern = [{"edge_type": "drug-to-protein", "direction": "out", "sample_size": 2}]
>>> graph_db.sample_typed_subgraph(["drug-1"], pattern)

save_manifest(path=None)

Write or update the root manifest for this graph.

Return type:

dict

serialize_edge_value(edge)

Serialize an edge for use with columnar edge ingestion.

Parameters:

edge (Edge)

Return type:

bytes

serialize_node_value(node)

Serialize a node for use with columnar node ingestion.

Parameters:

node (Node)

Return type:

bytes

stale_indexes()

Return index families requiring rebuild after deferred ingestion.

Return type:

tuple[str, …]

transaction(**options)

Run graph operations in a backend transaction when supported.

The transaction commits on clean context exit and rolls back if an exception leaves the context.

update_edge(edge_id, new_data, merge_func)

Similar approach for edges. The new_data might include new properties, or you might also allow changing source/target if that makes sense.

Parameters:

• edge_id (str)

• new_data (dict)

Return type:

Edge

update_node(node_id, new_data, merge_func)

Fetch existing node (if any). If none found, treat as new or handle gracefully. merge_func(old_node_dict, new_data_dict) -> merged_properties (dict)

Parameters:

• node_id (str)

• new_data (dict)

Return type:

Node

class gestaltdb.graphdb.GraphEntityDictSerializer(serializer)

Bases: object

Serialize graph entities through a dictionary-compatible serializer.

Parameters:

serializer (Serializer) – Serializer used for the final bytes conversion.

Examples

>>> from gestaltdb.serializers import JSONSerializer
>>> s = GraphEntityDictSerializer(JSONSerializer())
>>> s.deserialize(s.serialize(Node(node_id="n1"), "Node"), "Node").get_id
'n1'

__init__(serializer)

Initialize the entity serializer wrapper.

Parameters:

serializer (Serializer) – Serializer used to encode dictionaries as bytes.

deserialize(val, entity_type)

Deserializer (conditional on entity type)

Parameters:

• val – bytes containing the data

• entity_type (str) – (str) is Edge, Node, AdjacencyList

serialize(entity, entity_type)

Serialize a graph entity by entity type.

Parameters:

• entity – Node, Edge, or adjacency-list object.

• entity_type (str) – One of "Node", "Edge", or "AdjacencyList".

Returns:

Serialized bytes.

Examples

>>> from gestaltdb.serializers import PickleSerializer
>>> GraphEntityDictSerializer(PickleSerializer()).serialize(Node("n1"), "Node")[:1]
b'\x80'

class gestaltdb.graphdb.Node(node_id=None, properties=None, labels=None)

Bases: object

Graph node with an ID, native labels, and arbitrary properties.

Parameters:

• node_id – Optional stable node identifier. A UUID is generated when omitted.

• properties – Optional dictionary of node attributes.

• labels – Optional iterable of node labels. Labels are stored natively and maintained in the label index by GraphDB.

Examples

>>> Node(node_id="drug-1", labels=["Drug"], properties={"kind": "drug"}).get_id
'drug-1'
>>> Node(node_id="drug-1", labels=["Drug", "Drug"]).labels
('Drug',)

__init__(node_id=None, properties=None, labels=None)

Initialize a node, generating a UUID when node_id is omitted.

classmethod from_dict(data)

Create a node from serialized dictionary data.

Parameters:

data (dict) – Dictionary produced by to_dict. Older dictionaries without labels deserialize with an empty label tuple.

Returns:

Node instance.

Examples

>>> Node.from_dict({"id": "n1", "properties": {}}).labels
()

property get_id

Unique identifier for this node.

property get_id_bytes

Return the node ID encoded as UTF-8 bytes.

Examples

>>> Node(node_id="drug-1").get_id_bytes
b'drug-1'

to_dict()

Convert to a dictionary form for serialization.

Returns:

Dictionary containing id, properties, and labels.

Examples

>>> Node(node_id="n1", labels=["Drug"]).to_dict()["labels"]
['Drug']

class gestaltdb.graphdb.TimeIndexedEdge(timestamp_dat, *args, **kwargs)

Bases: Edge

Edge whose byte key is prefixed by a timestamp.

Parameters:

• timestamp_dat – Datetime used as the sortable key prefix.

• *args – Positional arguments passed to Edge.

• **kwargs – Keyword arguments passed to Edge.

Examples

>>> edge = TimeIndexedEdge(datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc), edge_id="e1")
>>> edge.get_id_bytes.endswith(b':e1')
True

__init__(timestamp_dat, *args, **kwargs)

Initialize a timestamp-prefixed edge.

Parameters:

• timestamp_dat – Datetime used as the sortable key prefix.

• *args – Positional arguments passed to Edge.

• **kwargs – Keyword arguments passed to Edge.

classmethod from_dict(data)

Factory from dictionary.

Parameters:

data (dict)

property get_id_bytes

Return timestamp-prefixed edge ID bytes.

Examples

>>> edge = TimeIndexedEdge(datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc), edge_id="e1")
>>> edge.get_id_bytes.endswith(b':e1')
True

to_dict()

Convert to a dictionary for serialization.

gestaltdb.graphdb.bytes_to_datetime(b, tzinfo=datetime.timezone.utc)

Convert bytes produced by datetime_to_bytes back to a datetime.

Parameters:

• b (bytes) – Eight-byte timestamp generated by datetime_to_bytes.

• tzinfo – Time zone used for the epoch reference.

Returns:

Decoded datetime.

Return type:

datetime

Examples

>>> bytes_to_datetime(b'\x00' * 8)
datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)

gestaltdb.graphdb.datetime_to_bytes(dt, tzinfo=datetime.timezone.utc)

Convert a datetime to big-endian microseconds since the Unix epoch.

Parameters:

• dt (datetime) – Datetime at or after 1970-01-01.

• tzinfo – Time zone used for the epoch reference.

Returns:

Eight bytes containing the timestamp as an unsigned integer.

Return type:

bytes

Examples

>>> datetime_to_bytes(datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc))
b'\x00\x00\x00\x00\x00\x00\x00\x00'

Sampling Configuration

Sampling APIs for GestaltDB.

This package preserves the original typed traversal configuration objects while adding array-native sampler snapshot and engine primitives.

class gestaltdb.sampling.AsyncBatchFeeder(producer, *, max_prefetch=2)

Bases: Generic[T]

Background producer that keeps a bounded queue of sampled batches.

Parameters:

• producer (Callable[[], T])

• max_prefetch (int)

get(timeout=None)

Return the next prefetched batch, raising producer errors inline.

Parameters:

timeout (float | None)

Return type:

T

start()

Start the background producer thread.

Return type:

AsyncBatchFeeder[T]

stop()

Signal the producer to stop and unblock waiting consumers.

Return type:

None

class gestaltdb.sampling.HardNegativeConfig(negatives_per_positive=1, source='random', reject_known_positives=True, same_endpoint_type=False, relation_endpoint_types=False, candidate_fanout=None, head_probability=0.5, max_retries=100)

Bases: object

Configuration for KG hard-negative generation.

Parameters:

• negatives_per_positive (int)

• source (str)

• reject_known_positives (bool)

• same_endpoint_type (bool)

• relation_endpoint_types (bool)

• candidate_fanout (int | None)

• head_probability (float)

• max_retries (int)

candidate_fanout: int | None

head_probability: float

max_retries: int

negatives_per_positive: int

reject_known_positives: bool

relation_endpoint_types: bool

same_endpoint_type: bool

source: str

class gestaltdb.sampling.SampledNeighbors(input_nodes, edge_indices, neighbor_nodes, offsets)

Bases: object

Batched neighbor-sampling result.

Parameters:

• input_nodes (ndarray)

• edge_indices (ndarray)

• neighbor_nodes (ndarray)

• offsets (ndarray)

input_nodes

Input compact node IDs, one per requested seed node.

Type:

numpy.ndarray

edge_indices

Sampled global compact edge IDs concatenated across all input nodes.

Type:

numpy.ndarray

neighbor_nodes

Neighbor compact node IDs aligned with edge_indices.

Type:

numpy.ndarray

offsets

Prefix-sum offsets into edge_indices and neighbor_nodes. Samples for input_nodes[i] are in offsets[i]:offsets[i + 1].

Type:

numpy.ndarray

Examples

Iterate over sampled neighbors for each input node:

result = engine.sample_neighbors([0, 1], fanout=5)
for row, node in enumerate(result.input_nodes):
    start, end = result.offsets[row], result.offsets[row + 1]
    neighbors = result.neighbor_nodes[start:end]

edge_indices: ndarray

input_nodes: ndarray

neighbor_nodes: ndarray

offsets: ndarray

class gestaltdb.sampling.SampledSubgraphBatch(node_ids_global, node_type_ids, senders, receivers, edge_ids_global, edge_relation_ids, positives, negatives, graph_node_offsets=None, graph_edge_offsets=None)

Bases: object

Framework-neutral sampled subgraph batch.

Node IDs inside senders, receivers, positives, and negatives are local to this batch. node_ids_global maps each local node row back to the compact global node ID in the SamplerSnapshot.

Parameters:

• node_ids_global (ndarray)

• node_type_ids (ndarray)

• senders (ndarray)

• receivers (ndarray)

• edge_ids_global (ndarray)

• edge_relation_ids (ndarray)

• positives (ndarray)

• negatives (ndarray)

• graph_node_offsets (ndarray | None)

• graph_edge_offsets (ndarray | None)

node_ids_global

Compact global node IDs included in this batch.

Type:

numpy.ndarray

node_type_ids

Node type IDs aligned with node_ids_global.

Type:

numpy.ndarray

senders

Local source node IDs for sampled edges.

Type:

numpy.ndarray

receivers

Local target node IDs for sampled edges.

Type:

numpy.ndarray

edge_ids_global

Compact global edge IDs included in this batch.

Type:

numpy.ndarray

edge_relation_ids

Relation IDs aligned with edge_ids_global.

Type:

numpy.ndarray

positives

Local positive triples shaped (num_positives, 3).

Type:

numpy.ndarray

negatives

Local negative triples, usually shaped (num_positives, negatives_per_positive, 3).

Type:

numpy.ndarray

graph_node_offsets

Optional graph component offsets for packed batches.

Type:

numpy.ndarray | None

graph_edge_offsets

Optional edge component offsets for packed batches.

Type:

numpy.ndarray | None

Examples

Convert to plain NumPy arrays:

arrays = batch.to_numpy()
edge_index = np.stack([arrays["senders"], arrays["receivers"]])

edge_ids_global: ndarray

edge_relation_ids: ndarray

graph_edge_offsets: ndarray | None

graph_node_offsets: ndarray | None

property n_edges: int

property n_nodes: int

negatives: ndarray

node_ids_global: ndarray

node_type_ids: ndarray

positives: ndarray

receivers: ndarray

senders: ndarray

to_arrow()

Return a PyArrow table for nodes and edges plus triple arrays.

Returns:

Dictionary with Arrow tables for nodes, edges, positives, and negatives. If negatives are grouped, a negative_group_offsets Arrow array is included.

Raises:

ImportError – If pyarrow is not installed.

Examples

Export node and edge tables:

arrow_batch = batch.to_arrow()
node_table = arrow_batch["nodes"]

to_dgl()

Return a DGL graph with node/edge features and label tensors.

Returns:

Tuple (graph, labels) where graph is a DGL graph and labels contains positive and negative PyTorch tensors.

Raises:

ImportError – If dgl or torch is not installed.

to_numpy()

Return arrays without framework-specific conversion.

Returns:

Dictionary containing NumPy arrays for nodes, edges, positives, and negatives. Optional graph offsets are included when present.

Return type:

dict[str, ndarray]

to_pyg()

Return a PyTorch Geometric-style dictionary of tensors.

Returns:

Dictionary with edge_index, node/edge type tensors, global ID tensors, and positive/negative label tensors.

Raises:

ImportError – If torch is not installed.

to_tf_gnns()

Return a TensorFlow-friendly graph dictionary and labels.

Returns:

Tuple (graph, labels) where both entries contain TensorFlow tensors. The graph dictionary follows the lightweight tensor-dict format used by the TarKG example.

Raises:

ImportError – If tensorflow is not installed.

class gestaltdb.sampling.SamplerEngine(snapshot, *, seed=None)

Bases: object

High-throughput sampler over a static SamplerSnapshot.

The engine samples compact integer arrays instead of materializing GraphDB Node/Edge objects. It supports direction-aware traversal, relation-aware traversal, multihop subgraph sampling, exact positive triple checks, and configurable hard-negative generation.

Parameters:

• snapshot (SamplerSnapshot) – Loaded sampler snapshot.

• seed (int | None) – Optional NumPy RNG seed for reproducible sampling.

Examples

Load a snapshot and sample two-hop neighborhoods around seed edges:

engine = SamplerEngine.load("data/sampler", mode="memmap", seed=7)
batch = engine.sample_subgraph([10, 11], fanouts=[15, 10])
arrays = batch.to_numpy()

is_positive(src, rel, dst)

Return whether a triple is a known positive in the snapshot.

Parameters:

• src (int) – Compact source node ID.

• rel (int) – Compact relation ID.

• dst (int) – Compact target node ID.

Returns:

True if (src, rel, dst) exists in the snapshot positives.

Return type:

bool

classmethod load(path, *, mode='ram', seed=None, **_kwargs)

Load a sampler engine from a snapshot path.

Parameters:

• path (str | Path) – Snapshot directory created by SamplerSnapshot.

• mode (str) – "ram" eagerly loads arrays; "memmap" memory maps them.

• seed (int | None) – Optional RNG seed.

• **_kwargs – Reserved for future cache/prefetch options.

Returns:

Initialized SamplerEngine.

Raises:

ValueError – If mode is not "ram" or "memmap".

Return type:

SamplerEngine

Examples

Load a RAM-backed engine:

engine = SamplerEngine.load("snapshot", mode="ram", seed=13)

sample_hard_negatives(positive_triples, context_nodes=None, config=None)

Generate grouped hard negatives for positive triples.

Parameters:

• positive_triples (Sequence[Sequence[int]] | ndarray) – Array-like positive triples shaped (n, 3) using compact global IDs.

• context_nodes (Iterable[int] | None) – Optional compact node IDs from the sampled context. Context-aware negative sources draw candidates from this set.

• config (HardNegativeConfig | None) – Negative sampling configuration. Defaults to HardNegativeConfig().

Returns:

Array shaped (num_positives, negatives_per_positive, 3) using compact global IDs.

Raises:

• ValueError – If positive_triples is not shaped (n, 3).

• RuntimeError – If a valid non-positive negative cannot be found within the configured retry budget.

Return type:

ndarray

sample_multihop(seeds, fanouts, *, direction='any', relations=None)

Sample a merged multihop subgraph around seed nodes.

Parameters:

• seeds (Sequence[int] | ndarray) – Compact node IDs used as the initial frontier.

• fanouts (Sequence[int]) – Fanout per hop. [15, 10] performs two hops.

• direction (str) – Traversal direction for all hops.

• relations (Sequence[int] | ndarray | None) – Optional relation filter applied to all hops.

Returns:

SampledSubgraphBatch containing one merged sampled graph.

Return type:

SampledSubgraphBatch

Examples

Sample a two-hop undirected context:

batch = engine.sample_multihop([u, v], [15, 10], direction="any")

sample_neighbors(nodes, fanout, *, direction='out', relations=None, replace=False)

Sample neighbors for each input node.

Parameters:

• nodes (Sequence[int] | ndarray) – Compact node IDs to sample from.

• fanout (int) – Maximum number of edges sampled per node.

• direction (str) – "out" for source-to-target, "in" for target-to-source, or "any" for incident traversal.

• relations (Sequence[int] | ndarray | None) – Optional compact relation IDs to restrict traversal.

• replace (bool) – Whether to sample with replacement.

Returns:

SampledNeighbors with concatenated edge and neighbor arrays.

Raises:

ValueError – If fanout is negative or direction is invalid.

Return type:

SampledNeighbors

Examples

Relation-aware neighbor sampling:

sampled = engine.sample_neighbors(
    [drug_id], fanout=10, direction="out", relations=[binds_id]
)

sample_subgraph(seed_edges, fanouts, *, direction='any', relations=None, negative_config=None)

Sample a merged training subgraph around positive seed edges.

Parameters:

• seed_edges (Sequence[int] | ndarray) – Compact edge IDs treated as positive triples.

• fanouts (Sequence[int]) – Fanout per hop around the positive edge endpoints.

• direction (str) – Traversal direction for context expansion.

• relations (Sequence[int] | ndarray | None) – Optional relation filter for context expansion.

• negative_config (HardNegativeConfig | None) – Optional hard-negative configuration. When provided, negatives are returned as (positives, negatives_per_positive, 3) local triples.

Returns:

SampledSubgraphBatch with local node IDs in positive and negative triples.

Return type:

SampledSubgraphBatch

Examples

Sample positives with hard negatives:

batch = engine.sample_subgraph(
    seed_edges,
    fanouts=[12, 8],
    negative_config=HardNegativeConfig(negatives_per_positive=8),
)

sample_subgraphs(*args, **kwargs)

Batched alias for sample_subgraph.

Return type:

SampledSubgraphBatch

class gestaltdb.sampling.SamplerSnapshot(path, metadata, external_node_ids, node_type_ids, external_edge_ids, external_relation_ids, relation_src_type_ids, relation_dst_type_ids, src_int, dst_int, rel_int, out, in_, incident, relation_out, relation_in, positive_triples)

Bases: object

Read-optimized array snapshot used by SamplerEngine.

A sampler snapshot is a derived, immutable index over a graph. It stores compact integer node IDs, compact integer relation IDs, edge endpoint arrays, CSR adjacency indexes, and exact positive triples. The durable graph database remains the source of truth; this object is optimized for training-time randomized sampling.

Parameters:

• path (Path)

• metadata (dict)

• external_node_ids (ndarray)

• node_type_ids (ndarray)

• external_edge_ids (ndarray)

• external_relation_ids (ndarray)

• relation_src_type_ids (ndarray)

• relation_dst_type_ids (ndarray)

• src_int (ndarray)

• dst_int (ndarray)

• rel_int (ndarray)

• out (CSRAdjacency)

• in_ (CSRAdjacency)

• incident (CSRAdjacency)

• relation_out (CSRAdjacency)

• relation_in (CSRAdjacency)

• positive_triples (ndarray)

path

Directory containing the persisted snapshot arrays and metadata.

Type:

pathlib.Path

metadata

JSON-compatible snapshot metadata.

Type:

dict

external_node_ids

External node identifiers ordered by compact node ID.

Type:

numpy.ndarray

node_type_ids

Integer node type ID per compact node, or -1 when unknown.

Type:

numpy.ndarray

external_edge_ids

External edge identifiers ordered by compact edge ID.

Type:

numpy.ndarray

external_relation_ids

External relation identifiers ordered by compact relation ID.

Type:

numpy.ndarray

relation_src_type_ids

Expected source node type ID per relation, or -1 when the source type is heterogeneous or unknown.

Type:

numpy.ndarray

relation_dst_type_ids

Expected target node type ID per relation, or -1 when the target type is heterogeneous or unknown.

Type:

numpy.ndarray

src_int

Source compact node ID for each edge.

Type:

numpy.ndarray

dst_int

Target compact node ID for each edge.

Type:

numpy.ndarray

rel_int

Relation compact ID for each edge.

Type:

numpy.ndarray

out

Source-to-edge CSR adjacency.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency

in_

Target-to-edge CSR adjacency.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency

incident

Undirected incident edge CSR adjacency.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency

relation_out

Relation-grouped source-to-edge CSR adjacency using keys node_id * num_relations + relation_id.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency

relation_in

Relation-grouped target-to-edge CSR adjacency using keys node_id * num_relations + relation_id.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency

positive_triples

Array of (src, rel, dst) positives.

Type:

numpy.ndarray

Examples

Build from already compact arrays and sample with SamplerEngine:

snapshot = SamplerSnapshot.from_edge_arrays(
    "snapshot",
    external_node_ids=["drug-1", "protein-1"],
    external_edge_ids=["edge-1"],
    external_relation_ids=["binds"],
    src_int=[0],
    dst_int=[1],
    rel_int=[0],
    node_type_values=["drug", "protein"],
)

classmethod build(graph, output_path, *, node_filter=None, edge_filter=None, edge_type_property='type', node_type_property='kind', directed=True, include_reverse=True, reverse_relation_policy='adjacency_only', storage='npy', layout='csr', source_db=None, source_artifacts=None)

Build and persist a static sampler snapshot from a GraphDB.

Parameters:

• graph – GraphDB-like object exposing store.get_node_keys_generator, store.get_edge_keys_generator, get_node, get_edge, and key_to_string.

• output_path (str | Path) – Directory where snapshot metadata and .npy arrays are written.

• node_filter (Callable[[object], bool] | None) – Optional predicate receiving each node object. Nodes for which the predicate returns False are omitted.

• edge_filter (Callable[[object], bool] | None) – Optional predicate receiving each edge object. Edges for which the predicate returns False are omitted.

• edge_type_property (str) – Edge property containing the relation identifier.

• node_type_property (str) – Node property containing the semantic node type.

• directed (bool) – Stored in metadata for consumers; current adjacency arrays preserve directed source/target endpoints and also build an incident view.

• include_reverse (bool) – Stored in metadata for consumers. Reverse synthetic relations are not currently materialized.

• reverse_relation_policy (str) – Metadata policy for reverse traversal. Must be "adjacency_only", "synthetic_relations", or "none".

• storage (str) – Metadata storage hint. Arrays are currently written as .npy files.

• layout (str) – Adjacency layout. Only "csr" is currently supported.

• source_db (dict | None)

• source_artifacts (dict | None)

Returns:

Loaded SamplerSnapshot pointing at output_path.

Raises:

ValueError – If reverse_relation_policy or layout is invalid.

Return type:

SamplerSnapshot

Examples

Build a snapshot from a graph database:

snapshot = graph.build_sampler_snapshot(
    "data/sampler",
    edge_type_property="type",
    node_type_property="kind",
)

dst_int: ndarray

external_edge_id(edge_int)

Return type:

str

external_edge_ids: ndarray

external_node_id(node_int)

Return type:

str

external_node_ids: ndarray

external_relation_id(rel_int)

Return type:

str

external_relation_ids: ndarray

classmethod from_arrays(output_path, *, external_node_ids, node_type_ids, external_edge_ids, external_relation_ids, src_int, dst_int, rel_int, relation_src_type_ids=None, relation_dst_type_ids=None, source_db=None, source_artifacts=None, metadata=None)

Build and persist a snapshot from fully encoded compact arrays.

Parameters:

• output_path (str | Path) – Directory where the snapshot is written.

• external_node_ids – External node IDs ordered by compact node ID.

• node_type_ids – Integer node type ID per compact node.

• external_edge_ids – External edge IDs ordered by compact edge ID.

• external_relation_ids – External relation IDs ordered by compact relation ID.

• src_int – Source compact node ID per edge.

• dst_int – Target compact node ID per edge.

• rel_int – Relation compact ID per edge.

• relation_src_type_ids – Optional source endpoint type ID per relation.

• relation_dst_type_ids – Optional target endpoint type ID per relation.

• metadata (dict | None) – Optional metadata merged into generated metadata.

• source_db (dict | None)

• source_artifacts (dict | None)

Returns:

Loaded SamplerSnapshot.

Return type:

SamplerSnapshot

Examples

Use this lower-level API when all type IDs are already encoded:

snapshot = SamplerSnapshot.from_arrays(
    "snapshot",
    external_node_ids=["n0", "n1"],
    node_type_ids=[0, 1],
    external_edge_ids=["e0"],
    external_relation_ids=["r0"],
    src_int=[0],
    dst_int=[1],
    rel_int=[0],
)

classmethod from_edge_arrays(output_path, *, external_node_ids, external_edge_ids, external_relation_ids, src_int, dst_int, rel_int, node_type_values=None, node_type_ids=None, edge_src_type_values=None, edge_dst_type_values=None, relation_src_type_ids=None, relation_dst_type_ids=None, source_db=None, source_artifacts=None, metadata=None)

Build and persist a snapshot from generic edge arrays.

This is the preferred API for pipelines that already have compact graph arrays, such as parquet/Arrow/NumPy preprocessing jobs. The method is deliberately free of pandas, Polars, Arrow, or database dependencies: the caller loads data with whichever tool is appropriate, then passes plain array-like values.

Exactly one of node_type_values or node_type_ids may be provided. If node_type_values is provided, values are deterministically encoded to integer IDs sorted by their string representation, with missing values encoded as -1. If edge endpoint type values are provided, they are used to derive relation-compatible source and target type constraints. Otherwise, relation endpoint type constraints are inferred from the node type IDs of the actual edge endpoints.

Parameters:

• output_path (str | Path) – Directory where the snapshot is written.

• external_node_ids (Sequence[object]) – External node IDs ordered by compact node ID.

• external_edge_ids (Sequence[object]) – External edge IDs ordered by compact edge ID.

• external_relation_ids (Sequence[object]) – External relation IDs ordered by compact relation ID.

• src_int – Source compact node ID per edge.

• dst_int – Target compact node ID per edge.

• rel_int – Relation compact ID per edge.

• node_type_values (Sequence[object] | None) – Optional semantic node type values ordered by compact node ID. Values are encoded to integer IDs.

• node_type_ids – Optional already encoded node type IDs ordered by compact node ID.

• edge_src_type_values (Sequence[object] | None) – Optional semantic source endpoint type value per edge, used to derive relation source endpoint constraints.

• edge_dst_type_values (Sequence[object] | None) – Optional semantic target endpoint type value per edge, used to derive relation target endpoint constraints.

• relation_src_type_ids – Optional precomputed relation source type IDs.

• relation_dst_type_ids – Optional precomputed relation target type IDs.

• metadata (dict | None) – Optional JSON-compatible metadata merged into the generated snapshot metadata.

• source_db (dict | None)

• source_artifacts (dict | None)

Returns:

Loaded SamplerSnapshot.

Raises:

ValueError – If array lengths are inconsistent, compact IDs are out of range, or both node_type_values and node_type_ids are provided.

Return type:

SamplerSnapshot

Examples

Build from NumPy arrays without any dataframe dependency:

snapshot = SamplerSnapshot.from_edge_arrays(
    "snapshot",
    external_node_ids=["n0", "n1", "n2"],
    external_edge_ids=["e0", "e1"],
    external_relation_ids=["r0"],
    src_int=np.array([0, 1]),
    dst_int=np.array([1, 2]),
    rel_int=np.array([0, 0]),
    node_type_values=["drug", "protein", "protein"],
)

get_edge(graph, edge_int)

get_node(graph, node_int)

global_triple_to_external(triple)

Return type:

tuple[str, str, str]

in_: CSRAdjacency

incident: CSRAdjacency

classmethod load(path, *, mmap=False)

Load a sampler snapshot from disk.

Parameters:

• path (str | Path) – Snapshot directory containing metadata.json and array files.

• mmap (bool) – If True, arrays are loaded using NumPy memory mapping.

Returns:

Loaded SamplerSnapshot.

Raises:

ValueError – If the snapshot format version is unsupported.

Return type:

SamplerSnapshot

Examples

Load eagerly into RAM:

snapshot = SamplerSnapshot.load("snapshot")

Load arrays through memory maps:

snapshot = SamplerSnapshot.load("snapshot", mmap=True)

local_triple_to_external(batch, triple)

Return type:

tuple[str, str, str]

metadata: dict

node_type_ids: ndarray

property num_edges: int

property num_nodes: int

property num_relations: int

open_source_graph(*, base_path=None, backend_options=None)

Open the GraphDB referenced by this snapshot’s source metadata.

out: CSRAdjacency

path: Path

positive_triples: ndarray

rel_int: ndarray

relation_dst_type_ids: ndarray

relation_in: CSRAdjacency

relation_out: CSRAdjacency

relation_src_type_ids: ndarray

source_graph_exists(*, base_path=None)

Return whether the referenced source graph path exists without opening it.

Return type:

bool

src_int: ndarray

class gestaltdb.sampling.SamplingHop(edge_type, direction='out', sample_size=10)

Bases: object

Configuration for one typed sampling hop.

Parameters:

• edge_type (str)

• direction (str)

• sample_size (int)

direction: str = 'out'

edge_type: str

classmethod from_dict(data)

Parameters:

data (Mapping[str, object])

Return type:

SamplingHop

sample_size: int = 10

to_dict()

Return type:

dict[str, object]

class gestaltdb.sampling.SamplingPattern(hops)

Bases: object

Ordered typed sampling pattern.

Parameters:

hops (Sequence[SamplingHop | Mapping[str, object]])

classmethod from_dicts(hops)

Parameters:

hops (Iterable[Mapping[str, object]])

Return type:

SamplingPattern

hops: Sequence[SamplingHop | Mapping[str, object]]

to_dicts()

Return type:

list[dict[str, object]]

gestaltdb.sampling.as_sampling_hop(hop)

Normalize a hop configuration to SamplingHop.

Parameters:

hop (SamplingHop | Mapping[str, object])

Return type:

SamplingHop

gestaltdb.sampling.as_sampling_pattern(pattern)

Normalize a sampling pattern to SamplingPattern.

Parameters:

pattern (SamplingPattern | Iterable[SamplingHop | Mapping[str, object]])

Return type:

SamplingPattern

Columnar Ingestion

Columnar ingestion containers for GestaltDB.

class gestaltdb.ingestion.ColumnarIngestionMode(*values)

Bases: str, Enum

How high-level columnar ingestion should interpret input columns.

ENTITY_COLUMNS means node and edge payloads are built from structured entity columns such as IDs, labels, sources, targets, types, and properties. With JSONSerializer, this enables the fast Polars/Arrow JSON payload path.

SERIALIZED_PAYLOADS means inputs already contain serializer-compatible node_value and edge_value byte payload columns.

ENTITY_COLUMNS = 'entity_columns'

SERIALIZED_PAYLOADS = 'serialized_payloads'

class gestaltdb.ingestion.EdgeList(edge_ids, sources, targets, edge_types, edge_values, edge_ids_column=None, sources_column=None, targets_column=None, edge_types_column=None, edge_values_column=None)

Bases: object

Columnar typed edges with caller-provided serialized edge values.

Parameters:

• edge_ids (list[bytes])

• sources (list[bytes])

• targets (list[bytes])

• edge_types (list[str])

• edge_values (list[bytes])

• edge_ids_column (object)

• sources_column (object)

• targets_column (object)

• edge_types_column (object)

• edge_values_column (object)

chunks(chunk_size)

Yield fixed-size EdgeList chunks.

Parameters:

chunk_size (int)

edge_ids: list[bytes]

edge_ids_column: object = None

edge_types: list[str]

edge_types_column: object = None

edge_values: list[bytes]

edge_values_column: object = None

classmethod from_arrow(edge_ids, sources, targets, edge_types, edge_values)

Create an edge list from Arrow-like or Python columns.

classmethod from_polars(df, *, edge_id='edge_id', source='source', target='target', edge_type='edge_type', edge_value='edge_value')

Create an edge list from a Polars DataFrame.

sources: list[bytes]

sources_column: object = None

targets: list[bytes]

targets_column: object = None

class gestaltdb.ingestion.IndexMaintenanceMode(*values)

Bases: str, Enum

Secondary-index maintenance policy for columnar ingestion.

MAINTAIN updates secondary indexes during ingestion. This is safest for incremental writes, but expensive for large append-only bulk loads.

DEFER writes canonical graph records and traversal-critical typed adjacency, marks secondary indexes stale, and requires an explicit GraphDB.rebuild_deferred_indexes() before index-backed queries.

DEFER_REBUILD is a high-level convenience mode: defer index maintenance during ingestion, then immediately run the one-pass deferred index rebuild.

DEFER = 'defer'

DEFER_REBUILD = 'defer_rebuild'

MAINTAIN = 'maintain'

class gestaltdb.ingestion.NodeList(node_ids, node_values, node_ids_column=None, node_values_column=None)

Bases: object

Columnar nodes with caller-provided serialized node values.

Parameters:

• node_ids (list[bytes])

• node_values (list[bytes])

• node_ids_column (object)

• node_values_column (object)

chunks(chunk_size)

Yield fixed-size NodeList chunks.

Parameters:

chunk_size (int)

classmethod from_arrow(node_ids, node_values)

Create a node list from Arrow-like or Python columns.

classmethod from_polars(df, *, node_id='node_id', node_value='node_value')

Create a node list from a Polars DataFrame.

node_ids: list[bytes]

node_ids_column: object = None

node_values: list[bytes]

node_values_column: object = None

Cypher Queries

Minimal read-only Cypher support for GestaltDB.

The supported subset maps directly to existing typed adjacency and sampling APIs:

MATCH (a {id: “node-id”})-[:TYPE1]->(b)<-[:TYPE2]-(c) RETURN a.name, b LIMIT 10 CALL pg.sample_typed_paths([“node-id”], [{“edge_type”: “TYPE”, “sample_size”: 2}]) YIELD path RETURN path

class gestaltdb.cypher.QueryResult(columns, records)

Bases: object

Tabular query result returned by GraphDB.query.

columns contains projected column names in return order. records is a list of dictionaries keyed by column name.

Examples

>>> result = QueryResult(columns=("n",), records=[{"n": "node"}])
>>> len(result)
1
>>> list(result)[0]["n"]
'node'

Parameters:

• columns (tuple[str, ...])

• records (list[dict[str, object]])

columns: tuple[str, ...]

records: list[dict[str, object]]

gestaltdb.cypher.execute(graph, query, parameters=None)

Execute a supported Cypher query against a GraphDB instance.

Parameters:

• graph – GraphDB instance used for indexed lookups and traversal.

• query (str) – Cypher query text.

• parameters (dict[str, object] | None)

Returns:

QueryResult with projected records.

Return type:

QueryResult

Examples

>>> execute(graph_db, 'MATCH (n:Drug) RETURN n')

gestaltdb.cypher.parse(query)

Parse the supported Cypher subset.

Parameters:

query (str) – Cypher query text.

Returns:

Parsed query object.

Raises:

ValueError – If the query is outside the supported subset.

Return type:

MatchQuery | SampleTypedPathsCall | NodeScanQuery | RelationshipScanQuery | MultiMatchQuery

Examples

>>> parse('MATCH (n:Drug) RETURN n').label
'Drug'

gestaltdb.cypher.plan(query)

Return the logical plan for a supported Cypher query.

Parameters:

query (str)

Return type:

LogicalPlan

Key-Value Stores

class gestaltdb.kvstores.KVStore

Bases: object

Abstract interface for a simple key-value store.

close()

Close any resources owned by the store.

delete(key)

Delete a raw key/value pair.

Parameters:

key (bytes)

delete_adjacency(node_id)

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)

Delete an edge.

Parameters:

edge_id (str)

delete_index_entry(index_name, key_parts, value)

Delete one sorted index entry.

Parameters:

• index_name (str) – Logical index name.

• key_parts (list[bytes]) – Ordered components used when the entry was written.

• value (bytes) – Entity ID or payload used when the entry was written.

Examples

>>> store.delete_index_entry("node_label", [b"Drug"], b"drug-1")

delete_metadata(key)

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)

Delete a node.

Parameters:

node_id (str)

delete_range_index_entry(index_name, key_parts, range_value, value)

Delete one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

delete_typed_adjacency(source_id, target_id, edge_type, edge_id)

Delete typed adjacency records for an edge.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

get(key)

Return a raw value by key.

Parameters:

key (bytes)

Return type:

bytes

get_edge(edge_id)

Retrieve an edge.

Parameters:

edge_id (str)

Return type:

bytes

get_edges_bulk(edge_ids)

Retrieve multiple serialized edges by key.

Parameters:

edge_ids (list[str])

Return type:

dict[str, bytes]

get_metadata(key)

Return a metadata value by key.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)

Retrieve a node by ID.

Parameters:

node_id (str)

Return type:

bytes

get_nodes_bulk(node_ids)

Retrieve multiple serialized nodes by key.

Parameters:

node_ids (list[str])

Return type:

dict[str, bytes]

ingest_edges_columnar(edge_list, *, append_only=True, native=True, maintain_indexes=True)

Store columnar typed edges with caller-provided serialized values.

Parameters:

• append_only (bool)

• native (bool)

• maintain_indexes (bool)

ingest_nodes_columnar(node_list, *, native=True)

Store columnar nodes with caller-provided serialized values.

Parameters:

native (bool)

iter_index_prefix(index_name, key_parts)

Yield values whose index key starts with key_parts.

Parameters:

• index_name (str) – Logical index name.

• key_parts (list[bytes]) – Ordered prefix components.

Yields:

Values associated with matching index entries.

Examples

>>> list(store.iter_index_prefix("node_label", [b"Drug"]))
[b'drug-1']

iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True)

Yield values whose range index key falls between start and end values.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• start_value (bytes | None)

• end_value (bytes | None)

• include_start (bool)

• include_end (bool)

iter_typed_adjacency(node_id, edge_type, direction='out')

Yield typed adjacency records for a node and edge type.

Parameters:

• node_id (bytes)

• edge_type (str)

• direction (str)

put(key, value)

Store a raw key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_edge(edge_id, value)

Store an edge (serialized).

Parameters:

• edge_id (str)

• value (bytes)

put_edges_bulk(keys_and_values)

Store multiple serialized edges.

Parameters:

keys_and_values (dict[str, bytes])

put_index_entries_bulk(entries)

Store many sorted index entries.

Parameters:

entries (list[tuple[str, list[bytes], bytes]]) – Tuples of (index_name, key_parts, value).

Examples

>>> store.put_index_entries_bulk([("node_label", [b"Drug"], b"drug-1")])

put_index_entry(index_name, key_parts, value)

Store one sorted index entry.

Parameters:

• index_name (str) – Logical index name, such as "node_label".

• key_parts (list[bytes]) – Ordered components used as the scan prefix.

• value (bytes) – Entity ID or payload returned by prefix scans.

Examples

>>> store.put_index_entry("node_label", [b"Drug"], b"drug-1")

put_metadata(key, value)

Store a metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_node(node_id, value)

Store a node (serialized).

Parameters:

• node_id (str)

• value (bytes)

put_nodes_bulk(keys_and_values)

Store multiple node (serialized) values in a single batch/transaction if possible.

Parameters:

keys_and_values (dict[str, bytes])

put_range_index_entries_bulk(entries)

Store many sorted range index entries.

Parameters:

entries (list[tuple[str, list[bytes], bytes, bytes]])

put_range_index_entry(index_name, key_parts, range_value, value)

Store one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

put_typed_adjacency(source_id, target_id, edge_type, edge_id)

Store typed adjacency records for an edge.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

put_typed_adjacency_bulk(records)

Store typed adjacency records for multiple edges.

Parameters:

records (list[tuple[bytes, bytes, str, bytes]])

range_iter(start_key, end_key)

Iterate over keys from start_key to end_key (inclusive).

Parameters:

• start_key (bytes)

• end_key (bytes)

supports_transactions = False

transaction(**options)

Return a transaction-bound store when supported.

class gestaltdb.kvstores.LMDBStore(path='graph_lmdb', map_size=10485760, map_id=True, map_keys=False)

Bases: KVStore

LMDB implementation of the GestaltDB key-value store.

Examples

>>> store = LMDBStore(path="/tmp/example_graph_lmdb")

__init__(path='graph_lmdb', map_size=10485760, map_id=True, map_keys=False)

Creates/opens an LMDB environment with three named sub-databases:

• b’nodes’ for node data

• b’edges’ for edge data

• b’adj’ for adjacency lists

close()

Close the LMDB environment.

delete(key)

Placeholder generic delete; graph code uses specialized methods.

Parameters:

key (bytes)

delete_adjacency(node_id)

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)

Delete an edge by byte key.

Parameters:

edge_id (str)

delete_index_entry(index_name, key_parts, value)

Delete one sorted index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• value (bytes)

delete_metadata(key)

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)

Delete a node by byte key.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)

Delete one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

delete_typed_adjacency(source_id, target_id, edge_type, edge_id)

Delete forward and reverse typed adjacency records.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

get(key)

Placeholder generic get; graph code uses specialized methods.

Parameters:

key (bytes)

Return type:

bytes

get_adjacency(node_id)

Return a serialized adjacency list for a node.

Parameters:

node_id (bytes | str)

Return type:

bytes | None

get_adjacency_bulk(node_ids)

Retrieve multiple adjacency lists in a single read transaction. Returns a dict { node_id: serialized adjacency } for all found items.

Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_edge(edge_id)

Return serialized edge bytes by key, or None.

Parameters:

edge_id (str)

Return type:

bytes

get_edge_keys_generator(num_edges=None, key_offset=None)

Yield edge keys from the edge database.

get_edges_bulk(edge_ids)

Return serialized edges for the requested keys.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)

Return a metadata value by key, or None.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)

Return serialized node bytes by key, or None.

Parameters:

node_id (bytes)

Return type:

bytes

get_node_keys_generator(num_nodes=None, key_offset=None)

Yield node keys from the node database.

get_nodes_bulk(node_ids)

Retrieve multiple nodes in one read transaction.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

iter_index_prefix(index_name, key_parts)

Yield values whose index key starts with key_parts.

Parameters:

• index_name (str)

• key_parts (list[bytes])

iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True)

Yield values whose range index key falls between start and end values.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• start_value (bytes | None)

• end_value (bytes | None)

• include_start (bool)

• include_end (bool)

iter_typed_adjacency(node_id, edge_type, direction='out')

Yield typed adjacency (edge_id, neighbor_id) pairs.

Parameters:

• node_id (bytes)

• edge_type (str)

• direction (str)

put(key, value)

Placeholder generic put; graph code uses specialized methods.

Parameters:

• key (bytes)

• value (bytes)

put_adjacency(node_id, value)

Store a serialized adjacency list for a node.

Parameters:

• node_id (bytes)

• value (bytes)

Return type:

None

put_adjacency_bulk(adj_dict)

Insert/update multiple adjacency lists in one transaction. :param adj_dict: a dict mapping node_id -> serialized adjacency (list of edges)

Parameters:

adj_dict (Dict[bytes, bytes])

Return type:

None

put_edge(edge_id, value)

Store a serialized edge by byte key.

Parameters:

• edge_id (bytes)

• value (bytes)

put_edges_bulk(keys_and_values)

Store many serialized edges in one transaction.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)

Store many sorted index entries in one transaction.

Parameters:

entries (list[tuple[str, list[bytes], bytes]])

put_index_entry(index_name, key_parts, value)

Store one sorted index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• value (bytes)

put_metadata(key, value)

Store a metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_node(node_id, value)

Store a serialized node by byte key.

Parameters:

• node_id (bytes)

• value (bytes)

put_nodes_bulk(keys_and_values)

Write a batch of nodes in a single transaction.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)

Store many sorted range index entries in one transaction.

Parameters:

entries (list[tuple[str, list[bytes], bytes, bytes]])

put_range_index_entry(index_name, key_parts, range_value, value)

Store one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

put_typed_adjacency(source_id, target_id, edge_type, edge_id)

Store forward and reverse typed adjacency records.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

put_typed_adjacency_bulk(records)

Store many typed adjacency records in one transaction.

Parameters:

records (list[tuple[bytes, bytes, str, bytes]])

range_iter(start_key, end_key)

Yield node records whose keys fall within an inclusive range.

Parameters:

• start_key (bytes)

• end_key (bytes)

supports_transactions = True

transaction(write=True, **options)

Open a transaction spanning all LMDB named databases.

Parameters:

write (bool)

class gestaltdb.kvstores.LMDBTransactionStore(parent, write=True, **options)

Bases: KVStore

Transaction-bound LMDB store using one environment transaction.

Parameters:

• parent (LMDBStore)

• write (bool)

close()

Close any resources owned by the store.

commit()

delete(key)

Delete a raw key/value pair.

Parameters:

key (bytes)

delete_adjacency(node_id)

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)

Delete an edge.

Parameters:

edge_id (bytes)

delete_index_entry(index_name, key_parts, value)

Delete one sorted index entry.

Parameters:

• index_name (str) – Logical index name.

• key_parts (list[bytes]) – Ordered components used when the entry was written.

• value (bytes) – Entity ID or payload used when the entry was written.

Examples

>>> store.delete_index_entry("node_label", [b"Drug"], b"drug-1")

delete_metadata(key)

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)

Delete a node.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)

Delete one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

delete_typed_adjacency(source_id, target_id, edge_type, edge_id)

Delete typed adjacency records for an edge.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

get(key)

Return a raw value by key.

Parameters:

key (bytes)

Return type:

bytes

get_adjacency(node_id)

Parameters:

node_id (bytes)

Return type:

bytes | None

get_adjacency_bulk(node_ids)

Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_edge(edge_id)

Retrieve an edge.

Parameters:

edge_id (bytes)

Return type:

bytes

get_edge_keys_generator(num_edges=None, key_offset=None)

get_edges_bulk(edge_ids)

Retrieve multiple serialized edges by key.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)

Return a metadata value by key.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)

Retrieve a node by ID.

Parameters:

node_id (bytes)

Return type:

bytes

get_node_keys_generator(num_nodes=None, key_offset=None)

get_nodes_bulk(node_ids)

Retrieve multiple serialized nodes by key.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

iter_index_prefix(index_name, key_parts)

Yield values whose index key starts with key_parts.

Parameters:

• index_name (str) – Logical index name.

• key_parts (list[bytes]) – Ordered prefix components.

Yields:

Values associated with matching index entries.

Examples

>>> list(store.iter_index_prefix("node_label", [b"Drug"]))
[b'drug-1']

iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True)

Yield values whose range index key falls between start and end values.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• start_value (bytes | None)

• end_value (bytes | None)

• include_start (bool)

• include_end (bool)

iter_typed_adjacency(node_id, edge_type, direction='out')

Yield typed adjacency records for a node and edge type.

Parameters:

• node_id (bytes)

• edge_type (str)

• direction (str)

put(key, value)

Store a raw key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_adjacency(node_id, value)

Parameters:

• node_id (bytes)

• value (bytes)

Return type:

None

put_adjacency_bulk(adj_dict)

Parameters:

adj_dict (Dict[bytes, bytes])

Return type:

None

put_edge(edge_id, value)

Store an edge (serialized).

Parameters:

• edge_id (bytes)

• value (bytes)

put_edges_bulk(keys_and_values)

Store multiple serialized edges.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)

Store many sorted index entries.

Parameters:

entries (list[tuple[str, list[bytes], bytes]]) – Tuples of (index_name, key_parts, value).

Examples

>>> store.put_index_entries_bulk([("node_label", [b"Drug"], b"drug-1")])

put_index_entry(index_name, key_parts, value)

Store one sorted index entry.

Parameters:

• index_name (str) – Logical index name, such as "node_label".

• key_parts (list[bytes]) – Ordered components used as the scan prefix.

• value (bytes) – Entity ID or payload returned by prefix scans.

Examples

>>> store.put_index_entry("node_label", [b"Drug"], b"drug-1")

put_metadata(key, value)

Store a metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_node(node_id, value)

Store a node (serialized).

Parameters:

• node_id (bytes)

• value (bytes)

put_nodes_bulk(keys_and_values)

Store multiple node (serialized) values in a single batch/transaction if possible.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)

Store many sorted range index entries.

Parameters:

entries (list[tuple[str, list[bytes], bytes, bytes]])

put_range_index_entry(index_name, key_parts, range_value, value)

Store one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

put_typed_adjacency(source_id, target_id, edge_type, edge_id)

Store typed adjacency records for an edge.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

put_typed_adjacency_bulk(records)

Store typed adjacency records for multiple edges.

Parameters:

records (list[tuple[bytes, bytes, str, bytes]])

range_iter(start_key, end_key)

Iterate over keys from start_key to end_key (inclusive).

Parameters:

• start_key (bytes)

• end_key (bytes)

rollback()

supports_transactions = True

transaction(**options)

Return a transaction-bound store when supported.

class gestaltdb.kvstores.LevelDBStore(path='graph_leveldb')

Bases: KVStore

LevelDB implementation backed by plyvel.

Parameters:

path – Directory that will contain the LevelDB sub-databases.

Examples

>>> store = LevelDBStore(path="/tmp/example_graph_leveldb")

__init__(path='graph_leveldb')

Create or open a LevelDB store. We’ll store nodes/edges by prefix.

close()

Close all LevelDB sub-databases.

delete_adjacency(node_id)

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)

Delete an edge by byte key.

Parameters:

edge_id (str)

delete_index_entry(index_name, key_parts, value)

Delete one sorted index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• value (bytes)

delete_metadata(key)

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)

Delete a node by byte key.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)

Delete one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

delete_typed_adjacency(source_id, target_id, edge_type, edge_id)

Delete forward and reverse typed adjacency records.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

get_adjacency(node_id)

Return a serialized adjacency list for a node.

Parameters:

node_id (bytes)

Return type:

bytes | None

get_adjacency_bulk(node_ids)

Return serialized adjacency lists for the requested nodes.

Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_db_iterator(which_db='nodes')

Yield all records from a named sub-database.

get_db_path(db_string='nodes')

Return the relative path for a named LevelDB database.

Examples

>>> LevelDBStore.get_db_path.__name__
'get_db_path'

get_edge(edge_id)

Return serialized edge bytes by key, or None.

Parameters:

edge_id (bytes)

Return type:

bytes

get_edge_keys_generator(num_edges=None, key_offset=None)

Yield edge keys from the edge database.

get_edges_bulk(edge_ids)

Return serialized edges for the requested keys.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)

Return a metadata value by key, or None.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)

Return serialized node bytes by key, or None.

Parameters:

node_id (bytes)

Return type:

bytes

get_node_keys_generator(num_nodes=None, key_offset=None)

Yield node keys from the node database.

get_node_keys_iterator()

Yield node database records.

get_nodes_bulk(node_ids)

Return serialized nodes for the requested keys.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

iter_index_prefix(index_name, key_parts)

Yield values whose index key starts with key_parts.

Parameters:

• index_name (str)

• key_parts (list[bytes])

iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True)

Yield values whose range index key falls between start and end values.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• start_value (bytes | None)

• end_value (bytes | None)

• include_start (bool)

• include_end (bool)

iter_typed_adjacency(node_id, edge_type, direction='out')

Yield typed adjacency (edge_id, neighbor_id) pairs.

Parameters:

• node_id (bytes)

• edge_type (str)

• direction (str)

put_adjacency(node_id, value)

Store a serialized adjacency list for a node.

Parameters:

• node_id (bytes)

• value (bytes)

Return type:

None

put_adjacency_bulk(adj_dict)

Insert/update multiple adjacency lists in one write batch. :param adj_dict: a dict mapping node_id -> serialized adjacency

Parameters:

adj_dict (Dict[str, bytes])

Return type:

None

put_edge(edge_id, value)

Store a serialized edge by byte key.

Parameters:

• edge_id (bytes)

• value (bytes)

put_edges_bulk(keys_and_values)

Store many serialized edges in one write batch.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)

Store many sorted index entries in one write batch.

Parameters:

entries (list[tuple[str, list[bytes], bytes]])

put_index_entry(index_name, key_parts, value)

Store one sorted index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• value (bytes)

put_metadata(key, value)

Store a metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_node(node_id, value)

Store a serialized node by byte key.

Parameters:

• node_id (bytes)

• value (bytes)

put_nodes_bulk(keys_and_values)

Use a WriteBatch for atomic bulk updates.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)

Store many sorted range index entries in one write batch.

Parameters:

entries (list[tuple[str, list[bytes], bytes, bytes]])

put_range_index_entry(index_name, key_parts, range_value, value)

Store one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

put_typed_adjacency(source_id, target_id, edge_type, edge_id)

Store forward and reverse typed adjacency records.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

put_typed_adjacency_bulk(records)

Store many typed adjacency records in one write batch.

Parameters:

records (list[tuple[bytes, bytes, str, bytes]])

range_iter(start_key, end_key)

Yield records whose keys fall within a range.

Note

This generic iterator is not used by the main graph APIs.

Parameters:

• start_key (bytes)

• end_key (bytes)

class gestaltdb.kvstores.PyRexStore(path='graph_rocksdb', parallelism=None, max_background_jobs=None, write_buffer_size=None, bloom_bits_per_key=None, disable_wal=False, transactional=False, transaction_db_options=None)

Bases: KVStore

RocksDB implementation backed by pyrex-rocksdb.

PyRexStore uses one physical RocksDB database with prefixed keys instead of separate databases. This lets node, edge, adjacency, and typed adjacency records share RocksDB’s write path and makes it possible to benchmark RocksDB tuning options against the existing LevelDB backend.

Parameters:

• path – Directory for the RocksDB database.

• parallelism – Optional number of RocksDB background threads.

• max_background_jobs – Optional RocksDB background job limit.

• write_buffer_size – Optional write buffer size in bytes.

• bloom_bits_per_key – Optional block-based Bloom filter bits per key.

• disable_wal – Disable RocksDB’s write-ahead log for faster but less durable ingestion benchmarks.

Examples

>>> store = PyRexStore(path="/tmp/example_graph_rocksdb")

__init__(path='graph_rocksdb', parallelism=None, max_background_jobs=None, write_buffer_size=None, bloom_bits_per_key=None, disable_wal=False, transactional=False, transaction_db_options=None)

Open a PyRex/RocksDB store with optional tuning settings.

close()

Close the RocksDB database.

delete(key)

Delete a raw key/value pair.

Parameters:

key (bytes)

delete_adjacency(node_id)

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)

Delete an edge by byte key.

Parameters:

edge_id (bytes)

delete_index_entry(index_name, key_parts, value)

Delete one sorted index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• value (bytes)

delete_metadata(key)

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)

Delete a node by byte key.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)

Delete one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

delete_typed_adjacency(source_id, target_id, edge_type, edge_id)

Delete forward and reverse typed adjacency records.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

get(key)

Return a raw value by key.

Parameters:

key (bytes)

Return type:

bytes

get_adjacency(node_id)

Return a serialized adjacency list for a node.

Parameters:

node_id (bytes)

Return type:

bytes | None

get_adjacency_bulk(node_ids)

Return serialized adjacency lists for the requested nodes.

Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_edge(edge_id)

Return serialized edge bytes by key, or None.

Parameters:

edge_id (bytes)

Return type:

bytes

get_edge_keys_generator(num_edges=None, key_offset=None)

Yield edge keys from the shared RocksDB keyspace.

get_edges_bulk(edge_ids)

Return serialized edges for the requested keys.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)

Return a metadata value by key, or None.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)

Return serialized node bytes by key, or None.

Parameters:

node_id (bytes)

Return type:

bytes

get_node_keys_generator(num_nodes=None, key_offset=None)

Yield node keys from the shared RocksDB keyspace.

get_nodes_bulk(node_ids)

Return serialized nodes for the requested keys.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

has_native_columnar_ingestion()

Return whether this PyRex runtime exposes native columnar writes.

Return type:

bool

ingest_edges_columnar(edge_list, *, append_only=True, native=True, maintain_indexes=True)

Store columnar typed edges, using native PyRex ingestion when available.

Parameters:

• append_only (bool)

• native (bool)

• maintain_indexes (bool)

ingest_nodes_columnar(node_list, *, native=True)

Store columnar nodes, using native PyRex ingestion when available.

Parameters:

native (bool)

iter_index_prefix(index_name, key_parts)

Yield values whose index key starts with key_parts.

Parameters:

• index_name (str)

• key_parts (list[bytes])

iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True)

Yield values whose range index key falls between start and end values.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• start_value (bytes | None)

• end_value (bytes | None)

• include_start (bool)

• include_end (bool)

iter_typed_adjacency(node_id, edge_type, direction='out')

Yield typed adjacency (edge_id, neighbor_id) pairs.

Parameters:

• node_id (bytes)

• edge_type (str)

• direction (str)

put(key, value)

Store a raw key/value pair in the shared RocksDB keyspace.

Parameters:

• key (bytes)

• value (bytes)

put_adjacency(node_id, value)

Store a serialized adjacency list for a node.

Parameters:

• node_id (bytes)

• value (bytes)

Return type:

None

put_adjacency_bulk(adj_dict)

Store many serialized adjacency lists in one RocksDB write batch.

Parameters:

adj_dict (Dict[bytes, bytes])

Return type:

None

put_edge(edge_id, value)

Store a serialized edge by byte key.

Parameters:

• edge_id (bytes)

• value (bytes)

put_edges_bulk(keys_and_values)

Store many serialized edges in one RocksDB write batch.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)

Store many sorted index entries in one write batch.

Parameters:

entries (list[tuple[str, list[bytes], bytes]])

put_index_entry(index_name, key_parts, value)

Store one sorted index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• value (bytes)

put_metadata(key, value)

Store a metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_node(node_id, value)

Store a serialized node by byte key.

Parameters:

• node_id (bytes)

• value (bytes)

put_nodes_bulk(keys_and_values)

Store many serialized nodes in one RocksDB write batch.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)

Store many sorted range index entries in one write batch.

Parameters:

entries (list[tuple[str, list[bytes], bytes, bytes]])

put_range_index_entry(index_name, key_parts, range_value, value)

Store one sorted range index entry.

Parameters:

• index_name (str)

• key_parts (list[bytes])

• range_value (bytes)

• value (bytes)

put_typed_adjacency(source_id, target_id, edge_type, edge_id)

Store forward and reverse typed adjacency records.

Parameters:

• source_id (bytes)

• target_id (bytes)

• edge_type (str)

• edge_id (bytes)

put_typed_adjacency_bulk(records)

Store many typed adjacency records in one RocksDB write batch.

Parameters:

records (list[tuple[bytes, bytes, str, bytes]])

range_iter(start_key, end_key)

Yield raw records whose keys fall within an inclusive range.

Parameters:

• start_key (bytes)

• end_key (bytes)

transaction(transaction_options=None, **options)

Open a RocksDB transaction-bound store.

class gestaltdb.kvstores.PyRexTransactionStore(parent, txn)

Bases: PyRexStore

Transaction-bound PyRex/RocksDB store.

Parameters:

parent (PyRexStore)

close()

Close the RocksDB database.

commit()

has_native_columnar_ingestion()

Return whether this PyRex runtime exposes native columnar writes.

Return type:

bool

rollback()

supports_transactions = True

transaction(**options)

Open a RocksDB transaction-bound store.

class gestaltdb.kvstores.SimpleIndexCounterKVStore(dbenv=None, db_path=b'nodes')

Bases: object

This is to help with lowering storage requirements for edge and node keys, by casting them to long ints.

It makes use of the struct.pack and struct.unpack functions and a simple counter (also stored in the medatadata) to count the number of keys (and hence the index) already entered.

__init__(dbenv=None, db_path=b'nodes')

Initialize an index counter helper.

Parameters:

• dbenv – LMDB environment.

• db_path – Named LMDB database for the counter mapping.

decode_db_key(key)

Return the stored encoded key bytes for a user key.

encode_db_key(key)

If the key exists, it will return the existing key. if the key does not exist, it will add it to the KV store with a new increment, and return that.

get(key)

Read a counter metadata value by key.

get_num_keys()

Return the number of keys already assigned.

put(key, value)

Store a counter metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_num_keys(num_keys)

Persist the number of keys already assigned.

class gestaltdb.kvstores.SimpleKV(db_path)

Bases: object

Small LMDB-backed helper for metadata key/value access.

Parameters:

db_path – LMDB database handle or name used by transactions.

__init__(db_path)

Initialize the helper with an LMDB database path or handle.

decode_db_key(key)

Return the encoded database key for a user key.

encode_db_key(key)

If the key exists, it will return the existing key. if the key does not exist, it will add it to the KV store with a new increment, and return that.

get(key)

Read a metadata value by key.

get_num_keys()

Return the stored key counter.

put(key, value)

Write a metadata key/value pair.

Parameters:

• key (bytes)

• value (bytes)

put_num_keys(num_keys)

Store the key counter.

Serializers

class gestaltdb.serializers.JSONSerializer

Bases: Serializer

Uses JSON for serialization.

deserialize(data)

Deserialize JSON bytes.

Examples

>>> JSONSerializer().deserialize(b'{"a": 1}')
{'a': 1}

Parameters:

data (bytes)

Return type:

dict

serialize(obj)

Serialize a JSON-compatible object.

Examples

>>> JSONSerializer().serialize({"a": 1})
b'{"a": 1}'

Parameters:

obj (dict)

Return type:

bytes

class gestaltdb.serializers.MessagePackSerializer

Bases: Serializer

Uses MessagePack for serialization.

deserialize(data)

Deserialize MessagePack bytes.

Raises:

ImportError – If the optional msgpack package is missing.

Parameters:

data (bytes)

Return type:

dict

Examples

>>> MessagePackSerializer().deserialize(MessagePackSerializer().serialize({"a": 1}))
{'a': 1}

serialize(obj)

Serialize an object with MessagePack.

Raises:

ImportError – If the optional msgpack package is missing.

Parameters:

obj (dict)

Return type:

bytes

Examples

>>> MessagePackSerializer().deserialize(MessagePackSerializer().serialize({"a": 1}))
{'a': 1}

class gestaltdb.serializers.PickleSerializer

Bases: Serializer

Uses Python’s pickle for serialization.

deserialize(data)

Deserialize pickle bytes.

Examples

>>> PickleSerializer().deserialize(PickleSerializer().serialize({"a": 1}))
{'a': 1}

Parameters:

data (bytes)

Return type:

dict

serialize(obj)

Serialize an object with pickle.

Examples

>>> PickleSerializer().deserialize(PickleSerializer().serialize({"a": 1}))
{'a': 1}

Parameters:

obj (dict)

Return type:

bytes

class gestaltdb.serializers.ProtobufSerializer

Bases: Serializer

Uses google.protobuf Struct for JSON-like dictionaries.

Struct does not have native integer or bytes types. This serializer tags those values before encoding so Python dictionaries round-trip without losing them.

deserialize(data)

Deserialize protobuf Struct bytes.

Parameters:

data (bytes) – Protobuf binary payload.

Returns:

Decoded dictionary.

Raises:

ImportError – If the optional protobuf package is missing.

Return type:

dict

serialize(obj)

Serialize a JSON-like dictionary with protobuf Struct.

Parameters:

obj (dict) – Dictionary containing JSON-like values plus tagged ints/bytes.

Returns:

Protobuf binary payload.

Raises:

ImportError – If the optional protobuf package is missing.

Return type:

bytes

class gestaltdb.serializers.Serializer

Bases: object

Abstract base for serialization/deserialization.

deserialize(data)

Deserialize bytes into a dictionary-like object.

Parameters:

data (bytes) – Serialized bytes.

Returns:

Decoded object.

Return type:

dict

serialize(obj)

Serialize a dictionary-like object to bytes.

Parameters:

obj (dict) – Object to serialize.

Returns:

Serialized bytes.

Return type:

bytes