API Reference

Graph Models and Database

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

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)[source]

If no edge_id is provided, generate a UUID.

classmethod from_dict(data)[source]

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()[source]

Convert to a dictionary for serialization.

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

Bases: object

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

Parameters:
__init__(store, serializer, indexed_node_properties=None, indexed_edge_properties=None)[source]

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"])
assert_claim(*, subject, predicate, object, polarity, agent, source=None, confidence=None, world='default', provenance=None, object_kind=ClaimObjectKind.ENTITY, valid=None, valid_from=None, valid_to=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append an immutable sourced assertion or denial.

Return type:

ClaimVersion

assert_world_accessibility(*, agent, from_world, to_world, kind, valid=None, valid_from=None, valid_to=None, source=None, provenance=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append a positive temporal accessibility fact for one agent frame.

Return type:

ClaimVersion

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

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, *, temporal=False, system_time=None, time_bucket=None, source_db_reference=True, source_db_path=None, source_db_path_mode='relative', read_snapshot=None, **kwargs)[source]

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

Parameters:

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

  • temporal (bool) – Build temporal-history arrays from one pinned read view.

  • system_time – Optional system-time horizon for a temporal build.

  • time_bucket (str | None) – Temporal candidate-index bucket policy: None, "none", "hour", or "day".

  • read_snapshot (bool | None) – Require or disable a verifiable mutable-state backend snapshot. The default uses one automatically when supported.

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

  • source_db_reference (bool)

  • source_db_path_mode (str)

Returns:

A loaded SamplerSnapshot instance.

claim_status(subject, predicate, object, *, valid_time, object_kind=ClaimObjectKind.ENTITY, agent=None, source=<object object>, world=None, system_time=None, through_commit=None)[source]

Return supported/refuted/both/unknown under open-world semantics.

Return type:

ClaimStatus

close()[source]

Close the underlying key-value store.

Examples

>>> graph_db.close()
commit_versions(writes, *, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append one logical commit containing immutable node/edge versions.

Return type:

TemporalCommit

correct_claim(claim, *, supersedes_version_id, confidence=<object object>, provenance=<object object>, valid=None, valid_from=None, valid_to=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append corrected confidence/provenance for one claim identity.

Return type:

ClaimVersion

correct_edge_version(edge, *, supersedes_version_id, valid=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append a corrected edge payload while preserving the superseded version.

Parameters:

  • edge (Edge)

  • supersedes_version_id (str)

Return type:

EdgeVersion

correct_node_version(node, *, supersedes_version_id, valid=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append a corrected node payload while preserving the superseded version.

Parameters:

  • node (Node)

  • supersedes_version_id (str)

Return type:

NodeVersion

count_edges_by_property(property_name, value)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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

Parameters:

  • property_name (str)

  • include_start (bool)

  • include_end (bool)

Return type:

int

count_typed_adjacency(node_id, edge_type, direction='out')[source]

Count typed adjacency records without materializing traversal records.

Parameters:

  • node_id – Node ID as string or bytes.

  • edge_type (str) – Edge type to traverse.

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

Returns:

Number of typed adjacency records matching the node, edge type, and direction.

Return type:

int

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

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

Return type:

GraphDB

create_edge_property_index(property_name)[source]

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)[source]

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
create_rule(rule_id, *, when, then, version_id=None)[source]

Validate and append one immutable version of a positive Horn rule.

Return type:

RuleVersion

current_read_view(*, require_verifiable=False, capture_provenance=True)[source]

Yield current graph reads pinned to one backend snapshot.

Parameters:

  • require_verifiable (bool)

  • capture_provenance (bool)

property database_id: str

Return the stable UUID persisted with this database.

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

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)[source]

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)[source]

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)[source]

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')[source]

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)[source]

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)[source]

Return edges using a scalar property range index.

Parameters:

  • property_name (str)

  • include_start (bool)

  • include_end (bool)

edges_by_type(edge_type)[source]

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)[source]

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)[source]

Return edges using a composite type/property range index.

Parameters:

  • edge_type (str)

  • property_name (str)

  • include_start (bool)

  • include_end (bool)

entails(agent, proposition, mode, *, world, valid_time, system_time=None, through_commit=None, max_depth=4, max_states=1000)[source]

Evaluate a bounded modal proposition under one pinned temporal view.

explain_claim(claim_id, *, valid_time, system_time=None, max_depth=100, max_nodes=10000)[source]

Return a finite derivation graph for a visible historical claim.

Return type:

ClaimExplanation | None

get_adjacency_list(node_id, direction='forward', return_raw=False)[source]

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_claim_as_of(claim_id, *, valid_time, system_time=None, through_commit=None)[source]

Resolve one claim identity at valid time and an optional system horizon.

Return type:

ClaimVersion | None

get_claim_version(version_id, *, system_time=None, through_commit=None)[source]

Return a visible epistemic claim version by UUID.

Parameters:

version_id (str)

Return type:

ClaimVersion | None

get_edge(edge_id)[source]

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_edge_as_of(logical_id, *, valid_time, system_time=None, through_commit=None)[source]

Resolve a logical edge at valid time and an optional system horizon.

Return type:

EdgeVersion | None

get_edge_version(version_id, *, system_time=None, through_commit=None)[source]

Return a visible edge version by UUID using the temporal index.

Parameters:

version_id (str)

Return type:

EdgeVersion | None

get_node(node_id)[source]

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_as_of(logical_id, *, valid_time, system_time=None, through_commit=None)[source]

Resolve a logical node at valid time and an optional system horizon.

Return type:

NodeVersion | None

get_node_keys_generator(num_nodes=None, key_offset=None)[source]

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_node_version(version_id, *, system_time=None, through_commit=None)[source]

Return a visible node version by UUID using the temporal index.

Parameters:

version_id (str)

Return type:

NodeVersion | None

get_nodes(node_ids)[source]

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_rule(rule_id, *, system_time=None)[source]

Return the latest version of a named rule at a system-time horizon.

Return type:

RuleVersion | None

get_temporal_commit(commit_id, *, _validate_supersession=True)[source]

Return a fully validated visible temporal commit.

Parameters:

  • commit_id (int)

  • _validate_supersession (bool)

Return type:

TemporalCommit | None

get_typed_adjacency(node_id, edge_type, direction='out')[source]

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()[source]

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)[source]

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:
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)[source]

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)[source]

Ingest edge entity columns from Arrow-like columns.

Parameters:
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)[source]

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:
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)[source]

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:
ingest_nodes_arrow(node_ids, node_values, *, native=True, chunk_size=100000, append_only=False, index_mode='maintain', progress=False)[source]

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)[source]

Ingest node entity columns from Arrow-like columns.

Parameters:
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)[source]

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:
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)[source]

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:
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)[source]

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:
Return type:

dict[str, object]

iter_claim_versions(claim_id=None, *, system_time=None, through_commit=None)[source]

Iterate claim versions, optionally restricted to one deterministic claim ID.

iter_claims_as_of(*, valid_time, statement_id=None, subject=None, predicate=None, object=<object object>, object_kind=None, agent=None, source=<object object>, world=None, polarity=None, system_time=None, through_commit=None)[source]

Iterate visible claims matching indexed epistemic dimensions.

iter_edge_ids(num_edges=None, key_offset=None)[source]

Yield all canonical edge IDs from the backing store.

Parameters:

  • num_edges – Optional maximum number of edge IDs to yield.

  • key_offset – Optional starting key.

Yields:

Edge ID bytes in backend key order.

Examples

>>> list(graph_db.iter_edge_ids())
[b'd1-p1']
iter_edge_ids_by_property(property_name, value)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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

Parameters:

  • edge_type (str)

  • property_name (str)

  • include_start (bool)

  • include_end (bool)

iter_edge_versions(logical_id=None, *, system_time=None, through_commit=None)[source]

Iterate edge versions, using the logical-history index when filtered.

iter_edges_as_of(*, edge_type=None, direction='out', source=None, target=None, valid_time, system_time=None, through_commit=None)[source]

Iterate endpoint edges visible at valid time and a system horizon.

iter_node_ids_by_label(label)[source]

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)[source]

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)[source]

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)[source]

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)[source]

Yield node IDs from a scalar property range index.

Parameters:

  • property_name (str)

  • include_start (bool)

  • include_end (bool)

iter_node_versions(logical_id=None, *, system_time=None, through_commit=None)[source]

Iterate node versions, using the logical-history index when filtered.

iter_nodes_as_of(*, valid_time, system_time=None, through_commit=None)[source]

Iterate node versions visible at one valid and system-time point.

iter_rule_versions(rule_id=None, *, system_time=None)[source]

Iterate immutable rule versions in catalog order at a system-time horizon.

iter_temporal_commits(*, through_commit=None)[source]

Iterate visible commits in commit order, skipping incomplete commits.

iter_typed_adjacency(node_id, edge_type, direction='out')[source]

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)[source]

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'
list_temporal_orphans(*, through_commit=None)[source]

List durable allocation gaps and incomplete unpublished commits.

Return type:

TemporalOrphanReport

maintain_truth(*, as_of=None, system_time=None, world=None, max_iterations=100, max_derivations=10000, max_justifications=100000, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Incrementally reconcile derived claims with the current rule fixpoint.

The authoritative base excludes claims emitted by the rule engine, so a cycle cannot preserve itself after its final independent support is removed. If as_of is omitted, the valid time from the most recent rule run or maintenance pass is resumed.

Return type:

TruthMaintenanceResult

property manifest: dict

Return the loaded or generated manifest for this graph.

migrate_time_indexed_edges(*, delete_legacy=False, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append legacy TimeIndexedEdge records as temporal edge history.

Records are grouped by logical edge ID and ordered by timestamp. Each record is valid until the next timestamp for that ID; the final record has an open end. Existing equivalent assertions from an earlier run are skipped, making cleanup retryable after a completed or interrupted migration. Changed legacy input after a successful migration is rejected instead of creating overlapping history. When requested, legacy current-state records are deleted only after the temporal commit is visible.

Returns:

Newly appended edge versions in migration order.

Raises:

  • TemporalVersionError – If timestamps are duplicated or legacy input changed after an earlier migration.

  • TemporalCorruptionError – If a legacy record cannot be decoded or its timestamp-prefixed key does not match its payload.

Parameters:

delete_legacy (bool)

Return type:

tuple[EdgeVersion, …]

neighbors_by_edge_type(node_id, edge_type, direction='out')[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

Open a self-describing graph store from a directory.

Return type:

GraphDB

put_adjacency_list(node_id, edges_list)[source]

Stores the adjacency list for node_id.

Parameters:
put_edge(edge, update_adjacency=True)[source]

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_edge_version(edge, *, valid, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append an immutable edge assertion without changing the current graph.

Parameters:

edge (Edge)

Return type:

EdgeVersion

put_edges_bulk(edges, check_existing=True)[source]

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)[source]

Store a node.

Parameters:

node (Node) – Node to serialize and write.

Examples

>>> graph_db.put_node(Node(node_id="drug-1"))
put_node_version(node, *, valid, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append an immutable node assertion without changing the current graph.

Parameters:

node (Node)

Return type:

NodeVersion

put_nodes(nodes)[source]

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, *, read_snapshot=None)[source]

Execute a supported Cypher query.

Read queries run directly; queries with CREATE/SET/REMOVE run inside transaction() when the backend supports it and fall back to best-effort direct writes otherwise.

Parameters:

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

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

  • read_snapshot (bool | None) – Require or disable a unified mutable-state read snapshot. The default uses one automatically when supported.

Returns:

gestaltdb.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)[source]

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

Parameters:

property_name (str)

read_view(*, valid_time, system_time=None, through_commit=None)[source]

Yield a temporal read view pinned to one visible commit horizon.

rebuild_deferred_indexes()[source]

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)[source]

Rebuild requested edge secondary indexes in one edge scan.

Parameters:
Return type:

dict[str, int]

rebuild_edge_property_index(property_name)[source]

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()[source]

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)[source]

Rebuild requested node secondary indexes in one node scan.

Parameters:
Return type:

dict[str, int]

rebuild_node_property_index(property_name)[source]

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()[source]

Rebuild the relationship type catalog from stored edges.

Returns:

Number of typed edge records indexed.

Examples

>>> graph_db.rebuild_relationship_type_index()
20
rebuild_temporal_indexes(*, through_commit=None)[source]

Rebuild all derived temporal indexes from visible canonical history.

Return type:

dict[str, int]

rebuild_typed_adjacency()[source]

Rebuild typed adjacency indexes from stored edge records.

Returns:

Number of typed edges indexed.

Examples

>>> graph_db.rebuild_typed_adjacency()
reclaim_temporal_orphans(*, through_commit)[source]

Delete validated unpublished artifacts through an explicit commit horizon.

Parameters:

through_commit (int)

Return type:

TemporalOrphanReclaimResult

retract_claim(claim_id, *, valid=None, valid_from=None, valid_to=None, supersedes_version_id=None, reason=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append a bitemporal retraction without deleting claim history.

Return type:

ClaimVersion

retract_edge_version(logical_id, *, valid=None, valid_from=None, supersedes_version_id=None, reason=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append an edge retraction over an explicit or inherited interval.

Parameters:

logical_id (str)

Return type:

EdgeVersion

retract_node_version(logical_id, *, valid=None, valid_from=None, supersedes_version_id=None, reason=None, version_id=None, metadata=None, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Append a node retraction over an explicit or inherited interval.

Parameters:

logical_id (str)

Return type:

NodeVersion

run_rules(*, as_of, system_time=None, world=None, max_iterations=100, max_derivations=10000, max_justifications=100000, index_mode=IndexMaintenanceMode.MAINTAIN)[source]

Evaluate active rules semi-naively and persist a bounded fixpoint.

Only positive entity-object claims participate. Premises must share a world, and each conclusion receives their half-open validity intersection. No claims are written if any resource bound is exceeded.

Return type:

RuleRunResult

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

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)[source]

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)[source]

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)[source]

Write or update the root manifest for this graph.

Return type:

dict

serialize_edge_value(edge)[source]

Serialize an edge for use with columnar edge ingestion.

Parameters:

edge (Edge)

Return type:

bytes

serialize_node_value(node)[source]

Serialize a node for use with columnar node ingestion.

Parameters:

node (Node)

Return type:

bytes

set_node_constraints(constraints)[source]

Replace and persist the node constraint catalog.

Parameters:

constraints (list[dict[str, str]])

Return type:

None

stale_indexes()[source]

Return index families requiring rebuild after deferred ingestion.

Return type:

tuple[str, …]

transaction(**options)[source]

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)[source]

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

Parameters:
Return type:

Edge

update_node(node_id, new_data, merge_func)[source]

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:
Return type:

Node

visualize(cypher=None, *, seeds=None, pattern=None, parameters=None, options=None, rng=None)[source]

Build an offline interactive visualization of this graph.

The front-end bundle is prebuilt and packaged with the library, so this works with no JavaScript toolchain and no network access. The returned figure saves self-contained .html artifacts and renders inline in Jupyter.

Parameters:

  • cypher – Optional Cypher query whose matched entities are visualized (with query matches highlighted).

  • seeds – Optional seed node IDs for typed-subgraph sampling; requires pattern.

  • pattern – SamplingPattern or hop dicts used with seeds.

  • parameters – Optional Cypher parameters for cypher.

  • options – Optional gestaltdb.viz.api.VizOptions or mapping.

  • rng – Optional random generator for sampling mode.

Returns:

gestaltdb.viz.api.VizFigure.

Examples

>>> figure = graph_db.visualize('MATCH (a:Person) RETURN a LIMIT 25')
>>> figure.save("/tmp/people.html")
class gestaltdb.graphdb.GraphEntityDictSerializer(serializer)[source]

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)[source]

Initialize the entity serializer wrapper.

Parameters:

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

deserialize(val, entity_type)[source]

Deserializer (conditional on entity type)

Parameters:

  • val – bytes containing the data

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

serialize(entity, entity_type)[source]

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)[source]

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)[source]

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

classmethod from_dict(data)[source]

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()[source]

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)[source]

Bases: Edge

Deprecated edge whose byte key is prefixed by a timestamp.

This legacy current-state layout is not the immutable temporal storage model. Use GraphDB.migrate_time_indexed_edges() to append equivalent edge history before adopting temporal reads or snapshots.

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)[source]

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)[source]

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()[source]

Convert to a dictionary for serialization.

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

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)[source]

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'

Temporal Semantics

Canonical temporal values shared by future graph and sampling APIs.

This module defines storage-independent temporal semantics. It does not make GraphDB records or sampler snapshots temporal by itself.

class gestaltdb.temporal.TemporalContext(instant=None, interval=None)[source]

Bases: object

A point-in-time or finite-window temporal selection.

Parameters:
matches(value)[source]

Test a point event or validity interval against this selection.

Parameters:

value (TemporalInterval | TemporalInstant | datetime | int)

Return type:

bool

class gestaltdb.temporal.TemporalDate(value)[source]

Bases: object

A calendar date without time or timezone.

Parameters:

value (date)

class gestaltdb.temporal.TemporalDuration(total_microseconds)[source]

Bases: object

An exact signed duration represented as total microseconds.

Parameters:

total_microseconds (int)

class gestaltdb.temporal.TemporalInstant(epoch_microseconds)[source]

Bases: object

An exact UTC instant represented as signed epoch microseconds.

Parameters:

epoch_microseconds (int)

classmethod decode_sortable(value)[source]

Decode the eight-byte chronological encoding from encode_sortable.

Parameters:

value (bytes)

Return type:

TemporalInstant

encode_sortable()[source]

Encode the instant so lexicographic byte order is chronological.

Return type:

bytes

classmethod parse(value)[source]

Parse an extended ISO-8601 datetime with an explicit UTC offset.

Parameters:

value (str)

Return type:

TemporalInstant

class gestaltdb.temporal.TemporalInterval(start, end=None)[source]

Bases: object

A non-empty half-open interval [start, end).

end=None denotes positive infinity.

Parameters:
intersection(other)[source]

Return the non-empty intersection, or None for disjoint intervals.

Parameters:

other (TemporalInterval)

Return type:

TemporalInterval | None

class gestaltdb.temporal.TemporalLocalDateTime(value)[source]

Bases: object

A timezone-free local date and time.

Parameters:

value (datetime)

class gestaltdb.temporal.TemporalLocalTime(microseconds)[source]

Bases: object

A timezone-free time represented as microseconds since midnight.

Parameters:

microseconds (int)

class gestaltdb.temporal.TemporalTime(local_microseconds, offset_seconds)[source]

Bases: object

A fixed-offset time of day.

Parameters:

  • local_microseconds (int)

  • offset_seconds (int)

gestaltdb.temporal.as_temporal_context(value)[source]

Normalize a context or adapt an instant-like value to an as-of context.

Parameters:

value (TemporalContext | TemporalInstant | datetime | int)

Return type:

TemporalContext

gestaltdb.temporal.as_temporal_instant(value)[source]

Normalize an instant, aware datetime, or epoch-microsecond integer.

Parameters:

value (TemporalInstant | datetime | int)

Return type:

TemporalInstant

gestaltdb.temporal.as_temporal_interval(value)[source]

Normalize an interval or a (start, end) pair.

Parameters:

value (TemporalInterval | Tuple[TemporalInstant | datetime | int, TemporalInstant | datetime | int | None])

Return type:

TemporalInterval

gestaltdb.temporal.datetime_to_epoch_microseconds(value)[source]

Normalize an aware datetime to exact UTC epoch microseconds.

Parameters:

value (datetime)

Return type:

int

gestaltdb.temporal.epoch_microseconds_to_datetime(value)[source]

Convert epoch microseconds to a timezone-aware UTC datetime.

Parameters:

value (int)

Return type:

datetime

Temporal Version History

Immutable temporal version records for GestaltDB.

The models in this module describe append-only history. Current-state graph records, temporal indexes, and as-of traversal are separate concerns.

class gestaltdb.versioning.ClaimVersion(version_id, logical_id, valid, commit_id, commit_ordinal, system_time, operation, payload_hash, supersedes_version_id=None, reason=None, claim=None)[source]

Bases: TemporalVersionMetadata

An immutable metadata envelope with an optional epistemic claim.

Parameters:
class gestaltdb.versioning.ClaimVersionWrite(operation, logical_id, valid, claim=None, version_id=None, supersedes_version_id=None, reason=None)[source]

Bases: object

Validated input describing an epistemic claim-version operation.

Parameters:
class gestaltdb.versioning.EdgeVersion(version_id, logical_id, valid, commit_id, commit_ordinal, system_time, operation, payload_hash, supersedes_version_id=None, reason=None, edge=None)[source]

Bases: TemporalVersionMetadata

An immutable metadata envelope with an optional edge payload.

Parameters:
class gestaltdb.versioning.EdgeVersionWrite(operation, logical_id, valid, edge=None, version_id=None, supersedes_version_id=None, reason=None)[source]

Bases: object

Validated input describing an edge-version operation.

Parameters:
class gestaltdb.versioning.NodeVersion(version_id, logical_id, valid, commit_id, commit_ordinal, system_time, operation, payload_hash, supersedes_version_id=None, reason=None, node=None)[source]

Bases: TemporalVersionMetadata

An immutable metadata envelope with an optional node payload.

Parameters:
class gestaltdb.versioning.NodeVersionWrite(operation, logical_id, valid, node=None, version_id=None, supersedes_version_id=None, reason=None)[source]

Bases: object

Validated input describing a node-version operation.

Parameters:
class gestaltdb.versioning.TemporalCommit(commit_id, system_time, metadata, versions)[source]

Bases: object

One visible temporal commit and its ordered entity versions.

Parameters:
exception gestaltdb.versioning.TemporalCorruptionError[source]

Bases: TemporalVersionError

Raised when marker-backed temporal history fails integrity checks.

exception gestaltdb.versioning.TemporalVersionError[source]

Bases: ValueError

Base error for invalid temporal version operations.

class gestaltdb.versioning.TemporalVersionMetadata(version_id, logical_id, valid, commit_id, commit_ordinal, system_time, operation, payload_hash, supersedes_version_id=None, reason=None)[source]

Bases: object

Metadata common to node and edge versions.

Parameters:
class gestaltdb.versioning.VersionOperation(*values)[source]

Bases: str, Enum

Operation represented by an immutable entity version.

Epistemic Claims

Immutable epistemic claim values and open-world status semantics.

class gestaltdb.epistemic.Claim(subject, predicate, object, polarity, agent, source=None, confidence=None, world='default', provenance=<factory>, object_kind=ClaimObjectKind.ENTITY)[source]

Bases: object

A sourced assertion or denial, independent of temporal version metadata.

statement_id identifies only the subject-predicate-object proposition. claim_id additionally identifies the asserting context (agent, source, world, and polarity), allowing contradictory claims to coexist.

Parameters:
classmethod from_dict(value)[source]

Decode and authenticate a canonical claim payload.

Parameters:

value (Mapping[str, object])

Return type:

Claim

to_dict()[source]

Return the canonical JSON-compatible claim payload.

Return type:

dict[str, object]

class gestaltdb.epistemic.ClaimObjectKind(*values)[source]

Bases: str, Enum

How a claim object is interpreted.

class gestaltdb.epistemic.ClaimPolarity(*values)[source]

Bases: str, Enum

Whether an agent asserts or denies a proposition.

class gestaltdb.epistemic.ClaimStatus(*values)[source]

Bases: str, Enum

Open-world four-valued status of a proposition at a read view.

gestaltdb.epistemic.claim_statement_id(subject, predicate, object, *, object_kind=ClaimObjectKind.ENTITY)[source]

Return the deterministic identity of a subject-predicate-object proposition.

Parameters:
Return type:

str

Positive Horn Rules

Immutable positive Horn rules and rule-evaluation result values.

class gestaltdb.rules.ClaimExplanation(claim_id, root_version_id, valid_time, system_time, nodes, edges, truncated=False)[source]

Bases: object

A finite historical explanation rooted at one visible claim version.

Parameters:
class gestaltdb.rules.ExplanationEdge(source_id, target_id, relationship, premise_ordinal=None)[source]

Bases: object

A directed derivation or premise edge in an explanation graph.

Parameters:

  • source_id (str)

  • target_id (str)

  • relationship (str)

  • premise_ordinal (int | None)

class gestaltdb.rules.ExplanationNode(node_id, kind, value)[source]

Bases: object

One immutable claim or rule version in an explanation graph.

Parameters:
class gestaltdb.rules.RuleAtom(subject, predicate, object)[source]

Bases: object

One (subject, predicate, object) atom in a positive Horn rule.

Parameters:
exception gestaltdb.rules.RuleError[source]

Bases: TemporalVersionError

Base error for invalid rule catalogs and evaluations.

exception gestaltdb.rules.RuleEvaluationLimitError[source]

Bases: RuleError

Raised before persistence when a rule run exceeds a resource bound.

class gestaltdb.rules.RuleJustification(rule_version_id, premise_version_ids)[source]

Bases: object

A rule version and the ordered claim versions that satisfied its body.

Parameters:

  • rule_version_id (str)

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

class gestaltdb.rules.RuleRunResult(iterations, derived_count, justification_count, commit_id, versions)[source]

Bases: object

Summary of one bounded rule run and its persisted claim versions.

Parameters:

  • iterations (int)

  • derived_count (int)

  • justification_count (int)

  • commit_id (int | None)

  • versions (tuple[object, ...])

class gestaltdb.rules.RuleVersion(rule_id, version_id, when, then, system_time, catalog_ordinal)[source]

Bases: object

One immutable system-time version of a named safe positive Horn rule.

Parameters:
class gestaltdb.rules.TruthMaintenanceResult(input_commit_id, commit_id, asserted_count, corrected_count, retracted_count, support_count, versions)[source]

Bases: object

Summary of one bounded incremental truth-maintenance pass.

Parameters:

  • input_commit_id (int)

  • commit_id (int | None)

  • asserted_count (int)

  • corrected_count (int)

  • retracted_count (int)

  • support_count (int)

  • versions (tuple[object, ...])

Modal Evaluation

Bounded four-valued modal evaluation over temporal epistemic claims.

class gestaltdb.modal.AccessibilityKind(*values)[source]

Bases: str, Enum

Independent accessibility frames used by modal operators.

class gestaltdb.modal.ModalEntailmentResult(status, confidence, explanation)[source]

Bases: object

Four-valued result and deterministic finite evaluation evidence.

Parameters:
exception gestaltdb.modal.ModalError[source]

Bases: TemporalVersionError

Base error for invalid modal formulas and evaluations.

exception gestaltdb.modal.ModalEvaluationLimitError[source]

Bases: ModalError

Raised when a bounded modal evaluation exhausts a configured limit.

class gestaltdb.modal.ModalExpression(operator, subject=None, predicate=None, object=None, object_kind=ClaimObjectKind.ENTITY, agent=None, operands=())[source]

Bases: object

A validated finite atom, connective, or nested modal expression.

Mapping forms use subject/predicate/object for atoms, formula for unary operators, and operands for AND/OR. Every modal operator requires an explicit agent.

Parameters:
class gestaltdb.modal.ModalOperator(*values)[source]

Bases: str, Enum

Operators accepted by ModalExpression.

gestaltdb.modal.evaluate_modal(view, expression, *, world, max_depth=4, max_states=1000)[source]

Evaluate one formula under a pinned read view and explicit finite bounds.

Parameters:
Return type:

ModalEntailmentResult

Consistent Read Views

Consistent temporal read views and reproducible source provenance.

class gestaltdb.readview.CurrentGraphReadView(graph, provenance)[source]

Bases: object

Graph facade whose current-state reads share one backend snapshot.

Parameters:

provenance (CurrentReadProvenance | None)

class gestaltdb.readview.CurrentReadProvenance(database_id, snapshot_sequence, backend_name, backend_layout_version, serializer_name, serializer_format_version, backend_snapshot_kind, state_sha256, snapshot_kind='mutable_backend_snapshot', format_version=1)[source]

Bases: object

Identity of one verifiable mutable-state backend snapshot.

Parameters:

  • database_id (str)

  • snapshot_sequence (int)

  • backend_name (str)

  • backend_layout_version (int)

  • serializer_name (str)

  • serializer_format_version (int)

  • backend_snapshot_kind (str)

  • state_sha256 (str)

  • snapshot_kind (str)

  • format_version (int)

verify_snapshot_store(view)[source]

Verify a provenance-free view is pinned to the captured sequence.

Return type:

None

verify_source(graph)[source]

Verify that the mutable source has not advanced since capture.

Return type:

None

verify_view(view)[source]

Verify a newly opened current read view has this exact identity.

Return type:

None

class gestaltdb.readview.GraphReadView(graph, provenance)[source]

Bases: object

A fixed temporal commit horizon and default valid-time instant.

Parameters:

provenance (ReadViewProvenance)

build_sampler_snapshot(output_path, **kwargs)[source]

Build a sampler snapshot from this exact source view.

evaluate_modal(expression, *, world, max_depth=4, max_states=1000)[source]

Evaluate a bounded modal expression at this view’s fixed horizon.

exception gestaltdb.readview.ProvenanceMismatchError[source]

Bases: ValueError

Raised when a source graph does not match captured read provenance.

class gestaltdb.readview.ReadViewProvenance(database_id, commit_horizon, commit_system_time_us, commit_marker_sha256, visibility_sha256, valid_time_us, backend_name, backend_layout_version, serializer_name, serializer_format_version, snapshot_kind='commit_visibility', format_version=1)[source]

Bases: object

JSON-serializable identity of one immutable temporal graph view.

Parameters:

  • database_id (str)

  • commit_horizon (int)

  • commit_system_time_us (int | None)

  • commit_marker_sha256 (str | None)

  • visibility_sha256 (str)

  • valid_time_us (int)

  • backend_name (str)

  • backend_layout_version (int)

  • serializer_name (str)

  • serializer_format_version (int)

  • snapshot_kind (str)

  • format_version (int)

classmethod from_dict(value)[source]

Parse and authenticate a provenance mapping.

Parameters:

value (dict)

Return type:

ReadViewProvenance

to_dict()[source]

Return canonical JSON-compatible provenance including its token.

Return type:

dict

to_json()[source]

Serialize provenance deterministically.

Return type:

str

property token: str

Return a stable digest over all provenance fields.

verify_source(graph)[source]

Verify that graph still represents this source view.

Return type:

None

gestaltdb.readview.parse_read_provenance(value)[source]

Parse temporal or mutable provenance by its explicit snapshot kind.

Parameters:

value (dict)

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.SamplingHop(edge_type, direction='out', sample_size=10)[source]

Bases: object

Configuration for one typed sampling hop.

Parameters:

  • edge_type (str)

  • direction (str)

  • sample_size (int)

class gestaltdb.sampling.SamplingPattern(hops)[source]

Bases: object

Ordered typed sampling pattern.

Parameters:

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

gestaltdb.sampling.as_sampling_hop(hop)[source]

Normalize a hop configuration to SamplingHop.

Parameters:

hop (SamplingHop | Mapping[str, object])

Return type:

SamplingHop

gestaltdb.sampling.as_sampling_pattern(pattern)[source]

Normalize a sampling pattern to SamplingPattern.

Parameters:

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

Return type:

SamplingPattern

Sampling Engine

Array-native sampler engine.

class gestaltdb.sampling.engine.SampledNeighbors(input_nodes, edge_indices, neighbor_nodes, offsets, edge_version_ids=None, valid_from_us=None, valid_to_us=None, valid_to_open=None)[source]

Bases: object

Batched neighbor-sampling result.

Parameters:

  • input_nodes (ndarray)

  • edge_indices (ndarray)

  • neighbor_nodes (ndarray)

  • offsets (ndarray)

  • edge_version_ids (ndarray | None)

  • valid_from_us (ndarray | None)

  • valid_to_us (ndarray | None)

  • valid_to_open (ndarray | None)

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

edge_version_ids

Sampled edge-version IDs for temporal snapshots.

Type:

numpy.ndarray | None

valid_from_us

Inclusive valid starts aligned with sampled edges.

Type:

numpy.ndarray | None

valid_to_us

Exclusive valid ends, with zero for open intervals.

Type:

numpy.ndarray | None

valid_to_open

Open-end masks aligned with sampled edges.

Type:

numpy.ndarray | None

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]
class gestaltdb.sampling.engine.SamplerEngine(snapshot, *, seed=None)[source]

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, *, temporal=None, policy='any_time', window_policy='overlap')[source]

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.

  • temporal (TemporalContext | None)

  • policy (str)

  • window_policy (str)

Returns:

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

Return type:

bool

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

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, *, positive_times_us=None, temporal=None, return_diagnostics=False)[source]

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().

  • positive_times_us (Sequence[int] | ndarray | None) – Optional example instants aligned with positives. Required by example-time policies and by leakage-safe candidate filtering unless temporal provides the needed context.

  • temporal (TemporalContext | None) – Optional point/window used by temporal positive rejection.

  • return_diagnostics (bool) – Return (negatives, counters) when true.

Returns:

Array shaped (num_positives, negatives_per_positive, 3) using compact global IDs, optionally paired with rejection counters.

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 | tuple[ndarray, dict[str, int]]

sample_multihop(seeds, fanouts, *, direction='any', relations=None, temporal=None, window_policy='overlap', causal_policy='none')[source]

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.

  • temporal (TemporalContext | None) – Optional point or finite-window validity filter.

  • window_policy (str) – Window matching policy, "overlap" or "contained".

  • causal_policy (str) – Valid-start policy across each sampled path: "none", "nondecreasing", or "nonincreasing".

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, temporal=None, window_policy='overlap', causal_policy='none')[source]

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.

  • temporal (TemporalContext | None) – Optional point or finite-window validity filter. Requires a temporal format-v2 snapshot.

  • window_policy (str) – "overlap" selects intervals intersecting a window; "contained" selects intervals fully inside it.

  • causal_policy (str) – Accepted for API consistency. It constrains valid starts across hops in multihop and subgraph sampling.

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, temporal=None, window_policy='overlap', causal_policy='none')[source]

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.

  • temporal (TemporalContext | None) – Optional point or finite-window validity filter applied to seed and context edges.

  • window_policy (str) – Window matching policy, "overlap" or "contained".

  • causal_policy (str) – Valid-start policy propagated from each seed edge through context hops.

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)[source]

Batched alias for sample_subgraph.

Return type:

SampledSubgraphBatch

Sampling Snapshots

Sampler snapshot build/load support.

class gestaltdb.sampling.snapshot.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, edge_version_ids=None, valid_from_us=None, valid_to_us=None, valid_to_open=None, edge_system_time_us=None, edge_commit_ids=None, temporal_out=None, temporal_in=None, positive_history_triples=None, positive_history_indptr=None, positive_history_edge_indices=None, node_history_indptr=None, node_valid_from_us=None, node_valid_to_us=None, node_valid_to_open=None)[source]

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)

  • edge_version_ids (ndarray | None)

  • valid_from_us (ndarray | None)

  • valid_to_us (ndarray | None)

  • valid_to_open (ndarray | None)

  • edge_system_time_us (ndarray | None)

  • edge_commit_ids (ndarray | None)

  • temporal_out (CSRAdjacency | None)

  • temporal_in (CSRAdjacency | None)

  • positive_history_triples (ndarray | None)

  • positive_history_indptr (ndarray | None)

  • positive_history_edge_indices (ndarray | None)

  • node_history_indptr (ndarray | None)

  • node_valid_from_us (ndarray | None)

  • node_valid_to_us (ndarray | None)

  • node_valid_to_open (ndarray | None)

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

edge_version_ids

Temporal edge-version UUIDs aligned with edge arrays, or None for non-temporal and legacy v1 snapshots.

Type:

numpy.ndarray | None

valid_from_us

Inclusive valid-time starts in UTC epoch microseconds.

Type:

numpy.ndarray | None

valid_to_us

Exclusive valid-time ends; open ends use zero together with valid_to_open.

Type:

numpy.ndarray | None

temporal_out

Source/relation CSR ordered by valid start.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency | None

temporal_in

Target/relation CSR ordered by valid start.

Type:

gestaltdb.sampling.adjacency.CSRAdjacency | None

positive_history_triples

Unique temporal positive triples.

Type:

numpy.ndarray | None

positive_history_indptr

Group offsets into positive history edge rows.

Type:

numpy.ndarray | None

positive_history_edge_indices

Edge rows ordered by triple and valid start.

Type:

numpy.ndarray | None

node_history_indptr

Node offsets into availability intervals.

Type:

numpy.ndarray | None

node_valid_from_us

Node availability interval starts.

Type:

numpy.ndarray | None

node_valid_to_us

Node availability interval ends.

Type:

numpy.ndarray | None

node_valid_to_open

Node availability open-end masks.

Type:

numpy.ndarray | None

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, source_provenance=None)[source]

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)

  • source_provenance (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",
)
classmethod build_temporal(read_view, output_path, *, node_filter=None, edge_filter=None, edge_type_property='type', node_type_property='kind', time_bucket='day', source_db=None, **metadata_options)[source]

Build a deterministic temporal-history snapshot from one read view.

Parameters:
Return type:

SamplerSnapshot

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, edge_version_ids=None, valid_from_us=None, valid_to_us=None, valid_to_open=None, edge_system_time_us=None, edge_commit_ids=None, node_intervals=None, source_db=None, source_artifacts=None, metadata=None)[source]

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)[source]

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"],
)
classmethod load(path, *, mmap=False)[source]

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)
open_source_graph(*, base_path=None, backend_options=None)[source]

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

source_graph_exists(*, base_path=None)[source]

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

Return type:

bool

property temporal: bool

Return whether this snapshot carries validated temporal arrays.

temporal_candidates(node, relation, valid_time, *, direction='out')[source]

Return a start-time-pruned superset for a temporal point query.

No edge valid at valid_time is omitted, while edges starting after the selected bucket are pruned. SamplerEngine applies the exact half-open point or window predicate before random selection.

Parameters:
Return type:

ndarray

verify_source(graph)[source]

Verify provenance-bearing snapshots against a source graph.

Return type:

None

Sampled Batches

Framework-neutral sampled batch containers.

class gestaltdb.sampling.batch.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, edge_version_ids=None, valid_from_us=None, valid_to_us=None, valid_to_open=None, positive_time_us=None, negative_time_us=None, negative_diagnostics=None)[source]

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)

  • edge_version_ids (ndarray | None)

  • valid_from_us (ndarray | None)

  • valid_to_us (ndarray | None)

  • valid_to_open (ndarray | None)

  • positive_time_us (ndarray | None)

  • negative_time_us (ndarray | None)

  • negative_diagnostics (dict[str, int] | 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

edge_version_ids

Edge-version IDs aligned with edge_ids_global for temporal snapshots, otherwise None.

Type:

numpy.ndarray | None

valid_from_us

Inclusive valid starts aligned with sampled edges.

Type:

numpy.ndarray | None

valid_to_us

Exclusive valid ends, with zero for open intervals.

Type:

numpy.ndarray | None

valid_to_open

Open-end masks aligned with sampled edges.

Type:

numpy.ndarray | None

positive_time_us

Example times aligned with positive triples.

Type:

numpy.ndarray | None

negative_time_us

Example times aligned with grouped negatives.

Type:

numpy.ndarray | None

negative_diagnostics

Deterministic rejection and exhaustion counters.

Type:

dict[str, int] | None

Examples

Convert to plain NumPy arrays:

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

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()[source]

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. Temporal graphs keep string edge-version IDs on graph.edge_version_ids.

Raises:

ImportError – If dgl or torch is not installed.

to_numpy()[source]

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()[source]

Return a PyTorch Geometric-style dictionary of tensors.

Returns:

Dictionary with edge_index, node/edge type tensors, global ID tensors, positive/negative label tensors, and temporal edge-version IDs as string-array metadata when present.

Raises:

ImportError – If torch is not installed.

to_tf_gnns()[source]

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.

Sampling Negatives

Negative sampling configuration.

class gestaltdb.sampling.negatives.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, temporal_positive_policy='any_time', temporal_positive_window_policy='overlap', temporal_candidate_window_days=None, allow_future_candidates=False, exhaustion_policy='raise')[source]

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)

  • temporal_positive_policy (str)

  • temporal_positive_window_policy (str)

  • temporal_candidate_window_days (float | None)

  • allow_future_candidates (bool)

  • exhaustion_policy (str)

Sampling Feeders

Async prefetch utilities for sampler batches.

class gestaltdb.sampling.feeder.AsyncBatchFeeder(producer, *, max_prefetch=2)[source]

Bases: Generic[T]

Background producer that keeps a bounded queue of sampled batches.

Parameters:

  • producer (Callable[[], T])

  • max_prefetch (int)

get(timeout=None)[source]

Return the next prefetched batch, raising producer errors inline.

Parameters:

timeout (float | None)

Return type:

T

start()[source]

Start the background producer thread.

Return type:

AsyncBatchFeeder[T]

stop()[source]

Signal the producer to stop and unblock waiting consumers.

Return type:

None

Columnar Ingestion

Columnar ingestion containers for GestaltDB.

class gestaltdb.ingestion.ColumnarIngestionMode(*values)[source]

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)[source]

Bases: object

Columnar typed edges with caller-provided serialized edge values.

Parameters:
chunks(chunk_size)[source]

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)[source]

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')[source]

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)[source]

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)[source]

Bases: object

Columnar nodes with caller-provided serialized node values.

Parameters:
chunks(chunk_size)[source]

Yield fixed-size NodeList chunks.

Parameters:

chunk_size (int)

classmethod from_arrow(node_ids, node_values)[source]

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

classmethod from_polars(df, *, node_id='node_id', node_value='node_value')[source]

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

Canonical Cypher query engine API.

class gestaltdb.query_engine.cypher.QueryResult(columns, records)[source]

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]]
gestaltdb.query_engine.cypher.execute(graph, query, parameters=None, *, read_snapshot=None)[source]

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)

  • read_snapshot (bool | None)

Returns:

QueryResult with projected records.

Return type:

QueryResult

Examples

>>> execute(graph_db, 'MATCH (n:Drug) RETURN n')
gestaltdb.query_engine.cypher.parse(query)[source]

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 | EntailsCall | NodeScanQuery | RelationshipScanQuery | MultiMatchQuery

Examples

>>> parse('MATCH (n:Drug) RETURN n').label
'Drug'
gestaltdb.query_engine.cypher.parse_ast(query)[source]

Parse the supported Cypher subset into its canonical clause AST.

Parameters:

query (str)

Return type:

Query | SampleTypedPathsCall | EntailsCall | UnionQuery | CreateConstraint | DropConstraint | ShowConstraints | ShowIndexes

gestaltdb.query_engine.cypher.plan(query)[source]

Return the logical plan for a supported Cypher query.

Clause queries plan to the staged operator pipeline, UNION queries plan each branch independently, and only the sampling procedure call keeps its dedicated source-backed plan. Constraint commands are executed directly and cannot be planned.

Parameters:

query (str)

Return type:

LogicalPlan

Key-Value Stores

class gestaltdb.kvstores.KVStore[source]

Bases: object

Abstract interface for a simple key-value store.

close()[source]

Close any resources owned by the store.

count_typed_adjacency(node_id, edge_type, direction='out')[source]

Count typed adjacency records for a node and edge type.

Parameters:
Return type:

int

delete(key)[source]

Delete a raw key/value pair.

Parameters:

key (bytes)

delete_adjacency(node_id)[source]

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)[source]

Delete an edge.

Parameters:

edge_id (str)

delete_index_entry(index_name, key_parts, value)[source]

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)[source]

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)[source]

Delete a node.

Parameters:

node_id (str)

delete_range_index_entry(index_name, key_parts, range_value, value)[source]

Delete one sorted range index entry.

Parameters:
delete_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Delete typed adjacency records for an edge.

Parameters:
get(key)[source]

Return a raw value by key.

Parameters:

key (bytes)

Return type:

bytes

get_edge(edge_id)[source]

Retrieve an edge.

Parameters:

edge_id (str)

Return type:

bytes

get_edges_bulk(edge_ids)[source]

Retrieve multiple serialized edges by key.

Parameters:

edge_ids (list[str])

Return type:

dict[str, bytes]

get_metadata(key)[source]

Return a metadata value by key.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)[source]

Retrieve a node by ID.

Parameters:

node_id (str)

Return type:

bytes

get_nodes_bulk(node_ids)[source]

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)[source]

Store columnar typed edges with caller-provided serialized values.

Parameters:
ingest_nodes_columnar(node_list, *, native=True)[source]

Store columnar nodes with caller-provided serialized values.

Parameters:

native (bool)

iter_index_prefix(index_name, key_parts)[source]

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, *, reverse=False, limit=None)[source]

Yield bounded range-index values, optionally greatest-first.

Parameters:
iter_typed_adjacency(node_id, edge_type, direction='out')[source]

Yield typed adjacency records for a node and edge type.

Parameters:
put(key, value)[source]

Store a raw key/value pair.

Parameters:
put_edge(edge_id, value)[source]

Store an edge (serialized).

Parameters:
put_edges_bulk(keys_and_values)[source]

Store multiple serialized edges.

Parameters:

keys_and_values (dict[str, bytes])

put_index_entries_bulk(entries)[source]

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)[source]

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)[source]

Store a metadata key/value pair.

Parameters:
put_node(node_id, value)[source]

Store a node (serialized).

Parameters:
put_nodes_bulk(keys_and_values)[source]

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)[source]

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)[source]

Store one sorted range index entry.

Parameters:
put_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Store typed adjacency records for an edge.

Parameters:
put_typed_adjacency_bulk(records)[source]

Store typed adjacency records for multiple edges.

Parameters:

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

range_iter(start_key, end_key)[source]

Iterate over keys from start_key to end_key (inclusive).

Parameters:
read_snapshot()[source]

Return a context manager yielding a unified read-only store view.

supports_read_snapshots = False
supports_transactions = False
supports_verifiable_read_snapshots = False
transaction(**options)[source]

Return a transaction-bound store when supported.

class gestaltdb.kvstores.LMDBReadSnapshotStore(parent)[source]

Bases: LMDBTransactionStore

Read-only LMDB transaction with a verifiable MVCC identity.

Parameters:

parent (LMDBStore)

commit()[source]
delete(*args, **kwargs)

Delete a raw key/value pair.

delete_adjacency(*args, **kwargs)

Delete a serialized adjacency list for a node.

delete_edge(*args, **kwargs)

Delete an edge.

delete_index_entry(*args, **kwargs)

Delete one sorted index entry.

Parameters:

  • index_name – Logical index name.

  • key_parts – Ordered components used when the entry was written.

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

Examples

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

Delete a metadata key/value pair.

delete_node(*args, **kwargs)

Delete a node.

delete_range_index_entry(*args, **kwargs)

Delete one sorted range index entry.

delete_typed_adjacency(*args, **kwargs)

Delete typed adjacency records for an edge.

put(*args, **kwargs)

Store a raw key/value pair.

put_adjacency(*args, **kwargs)
put_adjacency_bulk(*args, **kwargs)
put_edge(*args, **kwargs)

Store an edge (serialized).

put_edges_bulk(*args, **kwargs)

Store multiple serialized edges.

put_index_entries_bulk(*args, **kwargs)

Store many sorted index entries.

Parameters:

entries – Tuples of (index_name, key_parts, value).

Examples

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

Store one sorted index entry.

Parameters:

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

  • key_parts – Ordered components used as the scan prefix.

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

Examples

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

Store a metadata key/value pair.

put_node(*args, **kwargs)

Store a node (serialized).

put_nodes_bulk(*args, **kwargs)

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

put_range_index_entries_bulk(*args, **kwargs)

Store many sorted range index entries.

put_range_index_entry(*args, **kwargs)

Store one sorted range index entry.

put_typed_adjacency(*args, **kwargs)

Store typed adjacency records for an edge.

put_typed_adjacency_bulk(*args, **kwargs)

Store typed adjacency records for multiple edges.

state_sha256()[source]

Digest every mutable namespace visible at this MVCC horizon.

Return type:

str

supports_read_snapshots = False
supports_transactions = False
supports_verifiable_read_snapshots = True
class gestaltdb.kvstores.LMDBStore(path='graph_lmdb', map_size=10485760, map_id=True, map_keys=False)[source]

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)[source]
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()[source]

Close the LMDB environment.

current_snapshot_sequence()[source]

Return the latest committed LMDB transaction ID.

Return type:

int

delete(key)[source]

Placeholder generic delete; graph code uses specialized methods.

Parameters:

key (bytes)

delete_adjacency(node_id)[source]

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)[source]

Delete an edge by byte key.

Parameters:

edge_id (str)

delete_index_entry(index_name, key_parts, value)[source]

Delete one sorted index entry.

Parameters:
delete_metadata(key)[source]

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)[source]

Delete a node by byte key.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)[source]

Delete one sorted range index entry.

Parameters:
delete_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Delete forward and reverse typed adjacency records.

Parameters:
static ensure_available()[source]

Import and return LMDB, or raise installation guidance.

get(key)[source]

Placeholder generic get; graph code uses specialized methods.

Parameters:

key (bytes)

Return type:

bytes

get_adjacency(node_id)[source]

Return a serialized adjacency list for a node.

Parameters:

node_id (bytes | str)

Return type:

bytes | None

get_adjacency_bulk(node_ids)[source]

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)[source]

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)[source]

Yield edge keys from the edge database.

get_edges_bulk(edge_ids)[source]

Return serialized edges for the requested keys.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)[source]

Return a metadata value by key, or None.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)[source]

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)[source]

Yield node keys from the node database.

get_nodes_bulk(node_ids)[source]

Retrieve multiple nodes in one read transaction.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

iter_index_prefix(index_name, key_parts)[source]

Yield values whose index key starts with key_parts.

Parameters:
iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True, *, reverse=False, limit=None)[source]

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

Parameters:
iter_typed_adjacency(node_id, edge_type, direction='out')[source]

Yield typed adjacency (edge_id, neighbor_id) pairs.

Parameters:
put(key, value)[source]

Placeholder generic put; graph code uses specialized methods.

Parameters:
put_adjacency(node_id, value)[source]

Store a serialized adjacency list for a node.

Parameters:
Return type:

None

put_adjacency_bulk(adj_dict)[source]

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)[source]

Store a serialized edge by byte key.

Parameters:
put_edges_bulk(keys_and_values)[source]

Store many serialized edges in one transaction.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)[source]

Store many sorted index entries in one transaction.

Parameters:

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

put_index_entry(index_name, key_parts, value)[source]

Store one sorted index entry.

Parameters:
put_metadata(key, value)[source]

Store a metadata key/value pair.

Parameters:
put_node(node_id, value)[source]

Store a serialized node by byte key.

Parameters:
put_nodes_bulk(keys_and_values)[source]

Write a batch of nodes in a single transaction.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)[source]

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)[source]

Store one sorted range index entry.

Parameters:
put_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Store forward and reverse typed adjacency records.

Parameters:
put_typed_adjacency_bulk(records)[source]

Store many typed adjacency records in one transaction.

Parameters:

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

range_iter(start_key, end_key)[source]

Yield node records whose keys fall within an inclusive range.

Parameters:
read_snapshot()[source]

Pin all named databases to one LMDB MVCC transaction.

supports_read_snapshots = True
supports_transactions = True
supports_verifiable_read_snapshots = True
transaction(write=True, **options)[source]

Open a transaction spanning all LMDB named databases.

Parameters:

write (bool)

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

Bases: KVStore

Transaction-bound LMDB store using one environment transaction.

Parameters:
close()[source]

Close any resources owned by the store.

commit()[source]
delete(key)[source]

Delete a raw key/value pair.

Parameters:

key (bytes)

delete_adjacency(node_id)[source]

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)[source]

Delete an edge.

Parameters:

edge_id (bytes)

delete_index_entry(index_name, key_parts, value)[source]

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)[source]

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)[source]

Delete a node.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)[source]

Delete one sorted range index entry.

Parameters:
delete_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Delete typed adjacency records for an edge.

Parameters:
get(key)[source]

Return a raw value by key.

Parameters:

key (bytes)

Return type:

bytes

get_adjacency(node_id)[source]
Parameters:

node_id (bytes)

Return type:

bytes | None

get_adjacency_bulk(node_ids)[source]
Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_edge(edge_id)[source]

Retrieve an edge.

Parameters:

edge_id (bytes)

Return type:

bytes

get_edge_keys_generator(num_edges=None, key_offset=None)[source]
get_edges_bulk(edge_ids)[source]

Retrieve multiple serialized edges by key.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)[source]

Return a metadata value by key.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)[source]

Retrieve a node by ID.

Parameters:

node_id (bytes)

Return type:

bytes

get_node_keys_generator(num_nodes=None, key_offset=None)[source]
get_nodes_bulk(node_ids)[source]

Retrieve multiple serialized nodes by key.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

iter_index_prefix(index_name, key_parts)[source]

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, *, reverse=False, limit=None)[source]

Yield bounded range-index values, optionally greatest-first.

Parameters:
iter_typed_adjacency(node_id, edge_type, direction='out')[source]

Yield typed adjacency records for a node and edge type.

Parameters:
put(key, value)[source]

Store a raw key/value pair.

Parameters:
put_adjacency(node_id, value)[source]
Parameters:
Return type:

None

put_adjacency_bulk(adj_dict)[source]
Parameters:

adj_dict (Dict[bytes, bytes])

Return type:

None

put_edge(edge_id, value)[source]

Store an edge (serialized).

Parameters:
put_edges_bulk(keys_and_values)[source]

Store multiple serialized edges.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)[source]

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)[source]

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)[source]

Store a metadata key/value pair.

Parameters:
put_node(node_id, value)[source]

Store a node (serialized).

Parameters:
put_nodes_bulk(keys_and_values)[source]

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)[source]

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)[source]

Store one sorted range index entry.

Parameters:
put_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Store typed adjacency records for an edge.

Parameters:
put_typed_adjacency_bulk(records)[source]

Store typed adjacency records for multiple edges.

Parameters:

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

range_iter(start_key, end_key)[source]

Iterate over keys from start_key to end_key (inclusive).

Parameters:
rollback()[source]
supports_transactions = True
transaction(**options)[source]

Return a transaction-bound store when supported.

class gestaltdb.kvstores.LevelDBStore(path='graph_leveldb')[source]

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')[source]

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

close()[source]

Close all LevelDB sub-databases.

delete_adjacency(node_id)[source]

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)[source]

Delete an edge by byte key.

Parameters:

edge_id (str)

delete_index_entry(index_name, key_parts, value)[source]

Delete one sorted index entry.

Parameters:
delete_metadata(key)[source]

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)[source]

Delete a node by byte key.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)[source]

Delete one sorted range index entry.

Parameters:
delete_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Delete forward and reverse typed adjacency records.

Parameters:
static ensure_available()[source]

Import and return Plyvel, or raise installation guidance.

get_adjacency(node_id)[source]

Return a serialized adjacency list for a node.

Parameters:

node_id (bytes)

Return type:

bytes | None

get_adjacency_bulk(node_ids)[source]

Return serialized adjacency lists for the requested nodes.

Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_db_iterator(which_db='nodes')[source]

Yield all records from a named sub-database.

get_db_path(db_string='nodes')[source]

Return the relative path for a named LevelDB database.

Examples

>>> LevelDBStore.get_db_path.__name__
'get_db_path'
get_edge(edge_id)[source]

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)[source]

Yield edge keys from the edge database.

get_edges_bulk(edge_ids)[source]

Return serialized edges for the requested keys.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)[source]

Return a metadata value by key, or None.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)[source]

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)[source]

Yield node keys from the node database.

get_node_keys_iterator()[source]

Yield node database records.

get_nodes_bulk(node_ids)[source]

Return serialized nodes for the requested keys.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

iter_index_prefix(index_name, key_parts)[source]

Yield values whose index key starts with key_parts.

Parameters:
iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True, *, reverse=False, limit=None)[source]

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

Parameters:
iter_typed_adjacency(node_id, edge_type, direction='out')[source]

Yield typed adjacency (edge_id, neighbor_id) pairs.

Parameters:
put_adjacency(node_id, value)[source]

Store a serialized adjacency list for a node.

Parameters:
Return type:

None

put_adjacency_bulk(adj_dict)[source]

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)[source]

Store a serialized edge by byte key.

Parameters:
put_edges_bulk(keys_and_values)[source]

Store many serialized edges in one write batch.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)[source]

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)[source]

Store one sorted index entry.

Parameters:
put_metadata(key, value)[source]

Store a metadata key/value pair.

Parameters:
put_node(node_id, value)[source]

Store a serialized node by byte key.

Parameters:
put_nodes_bulk(keys_and_values)[source]

Use a WriteBatch for atomic bulk updates.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)[source]

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)[source]

Store one sorted range index entry.

Parameters:
put_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Store forward and reverse typed adjacency records.

Parameters:
put_typed_adjacency_bulk(records)[source]

Store many typed adjacency records in one write batch.

Parameters:

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

range_iter(start_key, end_key)[source]

Yield records whose keys fall within a range.

Note

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

Parameters:
read_snapshot()[source]

Return a context manager yielding a unified read-only store view.

class gestaltdb.kvstores.PyRexReadSnapshotStore(parent, txn, *, sequence)[source]

Bases: PyRexTransactionStore

Read-only transaction-bound PyRex store.

Parameters:
commit()[source]
delete(*args, **kwargs)

Delete a raw key/value pair.

delete_adjacency(*args, **kwargs)

Delete a serialized adjacency list for a node.

delete_edge(*args, **kwargs)

Delete an edge by byte key.

delete_index_entry(*args, **kwargs)

Delete one sorted index entry.

delete_metadata(*args, **kwargs)

Delete a metadata key/value pair.

delete_node(*args, **kwargs)

Delete a node by byte key.

delete_range_index_entry(*args, **kwargs)

Delete one sorted range index entry.

delete_typed_adjacency(*args, **kwargs)

Delete forward and reverse typed adjacency records.

put(*args, **kwargs)

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

put_adjacency(*args, **kwargs)

Store a serialized adjacency list for a node.

put_adjacency_bulk(*args, **kwargs)

Store many serialized adjacency lists in one RocksDB write batch.

put_edge(*args, **kwargs)

Store a serialized edge by byte key.

put_edges_bulk(*args, **kwargs)

Store many serialized edges in one RocksDB write batch.

put_index_entries_bulk(*args, **kwargs)

Store many sorted index entries in one write batch.

put_index_entry(*args, **kwargs)

Store one sorted index entry.

put_metadata(*args, **kwargs)

Store a metadata key/value pair.

put_node(*args, **kwargs)

Store a serialized node by byte key.

put_nodes_bulk(*args, **kwargs)

Store many serialized nodes in one RocksDB write batch.

put_range_index_entries_bulk(*args, **kwargs)

Store many sorted range index entries in one write batch.

put_range_index_entry(*args, **kwargs)

Store one sorted range index entry.

put_typed_adjacency(*args, **kwargs)

Store forward and reverse typed adjacency records.

put_typed_adjacency_bulk(*args, **kwargs)

Store many typed adjacency records in one RocksDB write batch.

state_sha256()[source]
Return type:

str

supports_read_snapshots = False
supports_transactions = False
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)[source]

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)[source]

Open a PyRex/RocksDB store with optional tuning settings.

close()[source]

Close the RocksDB database.

count_typed_adjacency(node_id, edge_type, direction='out')[source]

Count typed adjacency records without constructing record dictionaries.

Parameters:
Return type:

int

current_snapshot_sequence()[source]
Return type:

int | None

delete(key)[source]

Delete a raw key/value pair.

Parameters:

key (bytes)

delete_adjacency(node_id)[source]

Delete a serialized adjacency list for a node.

Parameters:

node_id (bytes)

delete_edge(edge_id)[source]

Delete an edge by byte key.

Parameters:

edge_id (bytes)

delete_index_entry(index_name, key_parts, value)[source]

Delete one sorted index entry.

Parameters:
delete_metadata(key)[source]

Delete a metadata key/value pair.

Parameters:

key (bytes)

delete_node(node_id)[source]

Delete a node by byte key.

Parameters:

node_id (bytes)

delete_range_index_entry(index_name, key_parts, range_value, value)[source]

Delete one sorted range index entry.

Parameters:
delete_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Delete forward and reverse typed adjacency records.

Parameters:
static ensure_available()[source]

Import and return PyRex, or raise installation guidance.

get(key)[source]

Return a raw value by key.

Parameters:

key (bytes)

Return type:

bytes

get_adjacency(node_id)[source]

Return a serialized adjacency list for a node.

Parameters:

node_id (bytes)

Return type:

bytes | None

get_adjacency_bulk(node_ids)[source]

Return serialized adjacency lists for the requested nodes.

Parameters:

node_ids (List[bytes])

Return type:

Dict[bytes, bytes]

get_edge(edge_id)[source]

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)[source]

Yield edge keys from the shared RocksDB keyspace.

get_edges_bulk(edge_ids)[source]

Return serialized edges for the requested keys.

Parameters:

edge_ids (list[bytes])

Return type:

dict[bytes, bytes]

get_metadata(key)[source]

Return a metadata value by key, or None.

Parameters:

key (bytes)

Return type:

bytes

get_node(node_id)[source]

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)[source]

Yield node keys from the shared RocksDB keyspace.

get_nodes_bulk(node_ids)[source]

Return serialized nodes for the requested keys.

Parameters:

node_ids (list[bytes])

Return type:

dict[bytes, bytes]

has_native_columnar_ingestion()[source]

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)[source]

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

Parameters:
ingest_nodes_columnar(node_list, *, native=True)[source]

Store columnar nodes, using native PyRex ingestion when available.

Parameters:

native (bool)

iter_index_prefix(index_name, key_parts)[source]

Yield values whose index key starts with key_parts.

Parameters:
iter_range_index(index_name, key_parts, start_value=None, end_value=None, include_start=True, include_end=True, *, reverse=False, limit=None)[source]

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

Parameters:
iter_typed_adjacency(node_id, edge_type, direction='out')[source]

Yield typed adjacency (edge_id, neighbor_id) pairs.

Parameters:
put(key, value)[source]

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

Parameters:
put_adjacency(node_id, value)[source]

Store a serialized adjacency list for a node.

Parameters:
Return type:

None

put_adjacency_bulk(adj_dict)[source]

Store many serialized adjacency lists in one RocksDB write batch.

Parameters:

adj_dict (Dict[bytes, bytes])

Return type:

None

put_edge(edge_id, value)[source]

Store a serialized edge by byte key.

Parameters:
put_edges_bulk(keys_and_values)[source]

Store many serialized edges in one RocksDB write batch.

Parameters:

keys_and_values (dict[bytes, bytes])

put_index_entries_bulk(entries)[source]

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)[source]

Store one sorted index entry.

Parameters:
put_metadata(key, value)[source]

Store a metadata key/value pair.

Parameters:
put_node(node_id, value)[source]

Store a serialized node by byte key.

Parameters:
put_nodes_bulk(keys_and_values)[source]

Store many serialized nodes in one RocksDB write batch.

Parameters:

keys_and_values (dict[bytes, bytes])

put_range_index_entries_bulk(entries)[source]

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)[source]

Store one sorted range index entry.

Parameters:
put_typed_adjacency(source_id, target_id, edge_type, edge_id)[source]

Store forward and reverse typed adjacency records.

Parameters:
put_typed_adjacency_bulk(records)[source]

Store many typed adjacency records in one RocksDB write batch.

Parameters:

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

range_iter(start_key, end_key)[source]

Yield raw records whose keys fall within an inclusive range.

Parameters:
read_snapshot()[source]

Pin reads to one RocksDB transaction snapshot when available.

transaction(transaction_options=None, **options)[source]

Open a RocksDB transaction-bound store.

class gestaltdb.kvstores.PyRexTransactionStore(parent, txn)[source]

Bases: PyRexStore

Transaction-bound PyRex/RocksDB store.

Parameters:

parent (PyRexStore)

close()[source]

Close the RocksDB database.

commit()[source]
has_native_columnar_ingestion()[source]

Return whether this PyRex runtime exposes native columnar writes.

Return type:

bool

rollback()[source]
supports_transactions = True
transaction(**options)[source]

Open a RocksDB transaction-bound store.

class gestaltdb.kvstores.ReadSnapshotIdentity(backend, kind, sequence, verifiable)[source]

Bases: object

Backend identity for one pinned mutable-state read horizon.

Parameters:

  • backend (str)

  • kind (str)

  • sequence (int | None)

  • verifiable (bool)

backend: str
kind: str
sequence: int | None
verifiable: bool
class gestaltdb.kvstores.SimpleIndexCounterKVStore(dbenv=None, db_path=b'nodes')[source]

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')[source]

Initialize an index counter helper.

Parameters:

  • dbenv – LMDB environment.

  • db_path – Named LMDB database for the counter mapping.

decode_db_key(key)[source]

Return the stored encoded key bytes for a user key.

encode_db_key(key)[source]

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)[source]

Read a counter metadata value by key.

get_num_keys()[source]

Return the number of keys already assigned.

put(key, value)[source]

Store a counter metadata key/value pair.

Parameters:
put_num_keys(num_keys)[source]

Persist the number of keys already assigned.

class gestaltdb.kvstores.SimpleKV(db_path)[source]

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)[source]

Initialize the helper with an LMDB database path or handle.

decode_db_key(key)[source]

Return the encoded database key for a user key.

encode_db_key(key)[source]

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)[source]

Read a metadata value by key.

get_num_keys()[source]

Return the stored key counter.

put(key, value)[source]

Write a metadata key/value pair.

Parameters:
put_num_keys(num_keys)[source]

Store the key counter.

Serializers

class gestaltdb.serializers.JSONSerializer[source]

Bases: Serializer

Uses JSON for serialization.

deserialize(data)[source]

Deserialize JSON bytes.

Examples

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

data (bytes)

Return type:

dict

serialize(obj)[source]

Serialize a JSON-compatible object.

Examples

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

obj (dict)

Return type:

bytes

class gestaltdb.serializers.MessagePackSerializer[source]

Bases: Serializer

Uses MessagePack for serialization.

deserialize(data)[source]

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}
static ensure_available()[source]

Import and return MessagePack, or raise installation guidance.

serialize(obj)[source]

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[source]

Bases: Serializer

Uses Python’s pickle for serialization.

deserialize(data)[source]

Deserialize pickle bytes.

Examples

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

data (bytes)

Return type:

dict

serialize(obj)[source]

Serialize an object with pickle.

Examples

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

obj (dict)

Return type:

bytes

class gestaltdb.serializers.ProtobufSerializer[source]

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)[source]

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

static ensure_available()[source]

Import and return protobuf helpers, or raise installation guidance.

serialize(obj)[source]

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[source]

Bases: object

Abstract base for serialization/deserialization.

deserialize(data)[source]

Deserialize bytes into a dictionary-like object.

Parameters:

data (bytes) – Serialized bytes.

Returns:

Decoded object.

Return type:

dict

serialize(obj)[source]

Serialize a dictionary-like object to bytes.

Parameters:

obj (dict) – Object to serialize.

Returns:

Serialized bytes.

Return type:

bytes

gestaltdb.serializers.temporal_tagged_value(value, *, canonical_time=False)[source]

Return a portable tagged representation for a temporal scalar.

Visualization IR

Deterministic visualization intermediate representation (VIZ-01).

Every visualization surface (static HTML files, Jupyter cells, client-side PNG/SVG/JSON export) renders the same JSON payload produced here. Builders accept Node/Edge objects (or their to_dict() shapes), Cypher QueryResult rows, sample_typed_subgraph mappings, and SampledSubgraphBatch + SamplerSnapshot pairs.

Standard library only: this module must not gain mandatory third-party runtime dependencies.

gestaltdb.viz.ir.DEFAULT_ABSOLUTE_CEILING_MULTIPLIER = 2

Multiplier applied to max_nodes/max_edges to derive the default absolute ceiling. Payloads beyond the ceiling raise instead of rendering.

gestaltdb.viz.ir.DEFAULT_MAX_EDGES = 5000

maximum edges kept in a payload before deterministic truncation.

Type:

Default cap

gestaltdb.viz.ir.DEFAULT_MAX_NODES = 2000

maximum nodes kept in a payload before deterministic truncation.

Type:

Default cap

gestaltdb.viz.ir.RESERVED_EXTRA_KEYS = ('valid', 'score', 'highlight')

Passthrough property names preserved verbatim on nodes/edges when present.

class gestaltdb.viz.ir.TruncationInfo(nodes_dropped=0, edges_dropped=0)[source]

Bases: object

Exact accounting of deterministic cap enforcement.

Parameters:

  • nodes_dropped (int)

  • edges_dropped (int)

edges_dropped: int = 0
nodes_dropped: int = 0
to_dict()[source]

Return the JSON-safe truncation payload.

Return type:

dict[str, Any]

property truncated: bool

Whether any node or edge was dropped.

exception gestaltdb.viz.ir.TruncationWarning[source]

Bases: UserWarning

Emitted when caps deterministically drop nodes or edges (VIZ-07).

gestaltdb.viz.ir.VIZ_IR_VERSION = 1

Version of the payload schema consumed by the web/ front end.

exception gestaltdb.viz.ir.VizCapExceededError[source]

Bases: ValueError

Raised when a payload exceeds its absolute ceiling (VIZ-07).

class gestaltdb.viz.ir.VizEdge(id, source, target, type='', properties=<factory>, extra=<factory>)[source]

Bases: object

A single directed edge in the visualization IR.

Parameters:
extra: dict[str, Any]
id: str
properties: dict[str, Any]
source: str
target: str
to_dict()[source]

Return the JSON-safe edge payload.

Return type:

dict[str, Any]

type: str = ''
class gestaltdb.viz.ir.VizGraph(nodes=(), edges=(), highlight_nodes=(), highlight_edges=(), truncation=<factory>, meta=<factory>)[source]

Bases: object

Deterministic, JSON-serializable graph payload for the front end.

Parameters:
property edge_ids: tuple[str, ...]

Edge IDs in payload order.

edges: tuple[VizEdge, ...] = ()
classmethod from_cypher_result(result, *, max_nodes=2000, max_edges=5000, absolute_max_nodes=None, absolute_max_edges=None, meta=None)[source]

Build a payload from a Cypher QueryResult.

Node/Edge/PathValue cell values become graph elements and populate the highlight set; node/edge-shaped dicts are accepted too. Scalar-only projections yield an empty topology and record their columns in meta so the outcome stays explainable.

Parameters:

  • result (Any)

  • max_nodes (int)

  • max_edges (int)

  • absolute_max_nodes (int | None)

  • absolute_max_edges (int | None)

  • meta (Mapping[str, Any] | None)

Return type:

VizGraph

classmethod from_nodes_edges(nodes, edges, *, max_nodes=2000, max_edges=5000, absolute_max_nodes=None, absolute_max_edges=None, highlight_nodes=(), highlight_edges=(), meta=None)[source]

Build a payload from node/edge objects or dict shapes.

Nodes sort by stable ID and keep the first max_nodes; edges sort by (source, target, id) and keep the first max_edges. Edges touching a dropped node are dropped as well and counted in edges_dropped. Duplicate IDs keep their first occurrence.

Payloads with more distinct nodes than absolute_max_nodes (or edges than absolute_max_edges) raise VizCapExceededError instead of freezing the browser; both default to twice their respective cap. Sample a subgraph or narrow the query instead.

Parameters:

  • nodes (Any)

  • edges (Any)

  • max_nodes (int)

  • max_edges (int)

  • absolute_max_nodes (int | None)

  • absolute_max_edges (int | None)

  • highlight_nodes (Any)

  • highlight_edges (Any)

  • meta (Mapping[str, Any] | None)

Return type:

VizGraph

classmethod from_sampled_subgraph(subgraph, *, max_nodes=2000, max_edges=5000, absolute_max_nodes=None, absolute_max_edges=None, meta=None)[source]

Build a payload from a sample_typed_subgraph mapping.

Missing (None) node/edge entries become ID-only placeholders so sampled paths never reference absent endpoints. Nodes and edges that appear on sampled paths populate the highlight set.

Parameters:
Return type:

VizGraph

classmethod from_sampler_batch(batch, snapshot, *, max_nodes=2000, max_edges=5000, absolute_max_nodes=None, absolute_max_edges=None, meta=None)[source]

Build a payload from a SampledSubgraphBatch and snapshot.

Local batch rows map back through node_ids_global to compact global IDs and then to external IDs via snapshot.external_node_id(...); relation IDs resolve via snapshot.external_relation_id(...). numpy is used only through duck typing (tolist/indexing), so this module keeps no mandatory third-party imports.

Parameters:

  • batch (Any)

  • snapshot (Any)

  • max_nodes (int)

  • max_edges (int)

  • absolute_max_nodes (int | None)

  • absolute_max_edges (int | None)

  • meta (Mapping[str, Any] | None)

Return type:

VizGraph

highlight_edges: tuple[str, ...] = ()
highlight_nodes: tuple[str, ...] = ()
meta: dict[str, Any]
property node_ids: tuple[str, ...]

Node IDs in payload order.

nodes: tuple[VizNode, ...] = ()
to_dict()[source]

Return the full JSON-safe payload consumed by the front end.

Return type:

dict[str, Any]

to_json()[source]

Return the canonical JSON encoding (sorted keys, compact).

Return type:

str

truncation: TruncationInfo
class gestaltdb.viz.ir.VizNode(id, labels=(), properties=<factory>, extra=<factory>)[source]

Bases: object

A single node in the visualization IR.

Parameters:
extra: dict[str, Any]
property group: str

Primary group used for coloring (first label, or "").

id: str
labels: tuple[str, ...] = ()
properties: dict[str, Any]
to_dict()[source]

Return the JSON-safe node payload.

Return type:

dict[str, Any]

gestaltdb.viz.ir.warn_if_truncated(viz_graph, *, source)[source]

Emit a TruncationWarning when caps dropped anything.

Parameters:

  • viz_graph (VizGraph) – Built payload to inspect.

  • source (str) – Human context (e.g. builder name) included in the message.

Return type:

None

Visualization Artifacts and API

Public Python API for packaged graph visualization (VIZ-06).

Thin conveniences over the IR (VIZ-01) and HTML (VIZ-03) layers. Standard library only; everything heavy stays behind function calls so import gestaltdb never pays for visualization.

class gestaltdb.viz.api.VizFigure(viz, options=<factory>)[source]

Bases: object

A built visualization: save it, embed it, or display it in Jupyter.

Parameters:
as_html()[source]

Return the self-contained offline HTML document.

Return type:

str

as_svg(*, width=1200, height=800, layout='circle', positions=None, background=True)[source]

Return dependency-free SVG using a static layout or supplied coordinates.

This renders the Python graph, not the current notebook iframe state. Set edge_labels='all' to include relationship labels in static output.

Parameters:
Return type:

str

options: VizOptions
save(path)[source]

Write the artifact to path and return the resolved path.

Parameters:

path (str | Path)

Return type:

Path

save_svg(path, **kwargs)[source]

Write as_svg() output and return the resolved destination.

Parameters:
Return type:

Path

viz: Any
class gestaltdb.viz.api.VizOptions(max_nodes=2000, max_edges=5000, absolute_max_nodes=None, absolute_max_edges=None, height=600, theme='light', charge=-300.0, link_distance=60.0, title='GestaltDB graph', show_labels=True, show_properties=True, node_label_property=None, edge_label_property=None, edge_labels='selected', label_max_length=40, node_size_property=None, node_color_property=None, edge_width_property=None, edge_color_property=None)[source]

Bases: object

Labels, property styling, caps, and initial notebook-view settings.

node_label_property / edge_label_property choose readable labels with ID/type fallbacks. edge_labels is off, selected (also hovered/focused), or all. Size/width properties use bounded numeric scales; color properties use categorical values. These mappings also apply to static SVG, where relationship labels require edge_labels='all'.

Parameters:

  • max_nodes (int)

  • max_edges (int)

  • absolute_max_nodes (int | None)

  • absolute_max_edges (int | None)

  • height (int)

  • theme (str)

  • charge (float)

  • link_distance (float)

  • title (str)

  • show_labels (bool)

  • show_properties (bool)

  • node_label_property (str | None)

  • edge_label_property (str | None)

  • edge_labels (str)

  • label_max_length (int)

  • node_size_property (str | None)

  • node_color_property (str | None)

  • edge_width_property (str | None)

  • edge_color_property (str | None)

absolute_max_edges: int | None = None
absolute_max_nodes: int | None = None
charge: float = -300.0
edge_color_property: str | None = None
edge_label_property: str | None = None
edge_labels: str = 'selected'
edge_width_property: str | None = None
height: int = 600
label_max_length: int = 40
max_edges: int = 5000
max_nodes: int = 2000
node_color_property: str | None = None
node_label_property: str | None = None
node_size_property: str | None = None
show_labels: bool = True
show_properties: bool = True
theme: str = 'light'
title: str = 'GestaltDB graph'
view_overrides()[source]

Return the front-end initial-state overrides for this options set.

Return type:

dict[str, Any]

gestaltdb.viz.api.visualize_nodes_edges(nodes, edges, *, options=None, **caps)[source]

Visualize explicit node/edge collections.

Parameters:

  • nodes (Any) – Node objects or node-shaped dicts.

  • edges (Any) – Edge objects or edge-shaped dicts.

  • options (VizOptions | Mapping[str, Any] | None) – VizOptions or equivalent mapping.

  • caps (Any) – Optional max_nodes/max_edges overrides.

Return type:

VizFigure

gestaltdb.viz.api.visualize_query(graph, cypher, parameters=None, *, options=None)[source]

Run a Cypher query and visualize the matched entities.

Parameters:
Return type:

VizFigure

gestaltdb.viz.api.visualize_sample(graph, seeds, pattern, *, rng=None, options=None)[source]

Sample a typed subgraph around seeds and visualize it.

Parameters:
Return type:

VizFigure

gestaltdb.viz.api.visualize_sampler_batch(batch, snapshot, *, options=None)[source]

Visualize an ML training batch mapped back to external IDs.

Local batch rows resolve through batch.node_ids_global to compact global IDs and then via snapshot.external_node_id(...) / snapshot.external_relation_id(...). The sampling-first path for inspecting SamplerEngine output without dumping whole snapshots.

Parameters:
Return type:

VizFigure

Self-contained offline HTML artifacts (VIZ-03).

Builds shareable .html files that inline the committed JS/CSS bundle plus the visualization payload. Standard library only: no new runtime dependency, no network, no template engine.

gestaltdb.viz.html.build_html(viz_graph, *, title='GestaltDB graph', warn_on_stale_bundle=True, view_overrides=None)[source]

Build a self-contained offline HTML document for a payload.

Parameters:

  • viz_graph (Any) – VizGraph (or any object exposing to_dict()) or an already-serialized payload mapping.

  • title (str) – Document title (HTML-escaped).

  • warn_on_stale_bundle (bool) – Emit a RuntimeWarning when the committed bundle does not match its manifest.

  • view_overrides (Mapping[str, Any] | None) – Optional front-end initial-state overrides (theme, charge, link distance, label/property visibility) injected as window.__GESTALTDB_VIZ_OPTIONS__.

Returns:

Complete HTML document as a string.

Return type:

str

gestaltdb.viz.html.ensure_bundle_fresh(*, warn=True)[source]

Warn (once per call) when the bundle does not match the manifest.

Returns:

stale output is still preferable to no output, but it must be loud.

Return type:

True when fresh, False otherwise. Never raises for staleness

Parameters:

warn (bool)

gestaltdb.viz.html.is_bundle_fresh()[source]

Verify asset hashes and, in a checkout, frontend source provenance.

Return type:

bool

gestaltdb.viz.html.read_manifest()[source]

Return the committed bundle manifest.

Raises:

FileNotFoundError – If the manifest is absent (e.g. source checkout where web/ was never built).

Return type:

dict[str, Any]

gestaltdb.viz.html.save_html(viz_graph, path, *, title='GestaltDB graph', warn_on_stale_bundle=True, view_overrides=None)[source]

Write a self-contained HTML artifact to path.

Creates parent directories as needed. Returns the resolved path.

Parameters:
Return type:

Path