TypeDB V2 store

Overview

The TypeDB V2 store allows you to store nodes using the Abstract store API. It maps canonical Node objects onto a TypeDB schema, providing graph database capabilities with strong typing and complex relationship modeling.

Configuration

Configuration of the typedb store happens using a dict containing an url, database and schema_path.

config = {
    "uri": "localhost:1729",
    "database": "knowledge_base",
    "schema_path": "path/to/schema.tql"  # Optional
}

Filter Logic and Practical Examples

The TypeDB V2 store uses a URL-query style filter syntax for retrieving and deleting nodes. This section explains how filters work with practical examples.

Filter Syntax

The basic filter format follows URL query string conventions:

entity=<entity_type>&<attribute>=<value>&<attribute>=<value>&include=relations

Practical Examples

Example 1: Single Attribute Filter

# Find a person by email
filter = "entity=person&email=john.doe@example.com"
nodes = store.get_nodes(filter)

# Find documents by title
filter = "entity=document&title=Research Paper 2024"
nodes = store.get_nodes(filter)

Example 2: Multiple Attribute Filters

# Find a person by email AND department
filter = "entity=person&email=jane@company.com&department=Engineering"
nodes = store.get_nodes(filter)

# Find documents by author AND status
filter = "entity=document&author=Dr. Smith&status=published"
nodes = store.get_nodes(filter)

Example 3: Including Relations

# Get person with all their relations loaded
filter = "entity=person&email=john@example.com&include=relations"
nodes = store.get_nodes(filter)

# The returned node will have its relations populated:
# node.relations will contain RelationData objects
for node in nodes:
    print(f"Found {len(node.relations)} relations for {node.id}")
    for rel in node.relations:
        print(f"  - {rel['type']} relation with roles: {rel['roles']}")

Example 4: Retrieving All Nodes

# Get all nodes in the database (use with caution on large datasets)
nodes = store.get_nodes(None)

# This is equivalent to:
nodes = store.get_nodes(filter=None)

Filter Behavior and Edge Cases

Empty vs None Filter

# None filter returns ALL nodes
nodes = store.get_nodes(None)  # Returns everything

# Empty string filter raises ValueError
nodes = store.get_nodes("")  # Raises ValueError: filter cannot be empty

Case Sensitivity

# Filters are case-sensitive
filter1 = "entity=person&name=John Doe"
filter2 = "entity=person&name=john doe"
# These will return different results

Special Characters in Values

# Values with special characters are handled automatically
filter = "entity=person&email=user+tag@example.com"
# The + character is preserved correctly

Deletion Filters

The same filter syntax applies to remove_nodes():

Safe Deletion (Single Entity)

# Delete a specific person
filter = "entity=person&email=john@example.com"
deleted_count = store.remove_nodes(filter)

Bulk Deletion (Requires Permission)

# Delete all draft documents (requires allow_multiple=True)
filter = "entity=document&status=draft"
deleted_count = store.remove_nodes(filter, allow_multiple=True)

# Attempting bulk deletion without permission raises ValueError
filter = "entity=document"  # No attribute filter
deleted_count = store.remove_nodes(filter)  # Raises ValueError

Docstring

typedb_store

Classes:

• EagerQueryAnswer –

  Eagerly evaluates a TypeDB QueryAnswer to prevent 'concurrent transaction close' errors

• TypeDbDatastore –

  TypeDB-backed implementation of the AbstractStore knowledge persistence layer.

EagerQueryAnswer

EagerQueryAnswer(answer: QueryAnswer)

Eagerly evaluates a TypeDB QueryAnswer to prevent 'concurrent transaction close' errors when evaluating iterator wrappers outside of the transaction block.

Source code in src/database_builder_libs/stores/typedb/typedb_store.py

def __init__(self, answer: QueryAnswer):
    self._is_docs = answer.is_concept_documents()
    self._is_rows = answer.is_concept_rows()
    if self._is_docs:
        self._docs = list(answer.as_concept_documents())
    elif self._is_rows:
        self._rows = list(answer.as_concept_rows())
    else:
        self._answer = answer

TypeDbDatastore

TypeDbDatastore()

              flowchart TD
              database_builder_libs.stores.typedb.typedb_store.TypeDbDatastore[TypeDbDatastore]
              database_builder_libs.models.abstract_store.AbstractStore[AbstractStore]

                              database_builder_libs.models.abstract_store.AbstractStore --> database_builder_libs.stores.typedb.typedb_store.TypeDbDatastore
                


              click database_builder_libs.stores.typedb.typedb_store.TypeDbDatastore href "" "database_builder_libs.stores.typedb.typedb_store.TypeDbDatastore"
              click database_builder_libs.models.abstract_store.AbstractStore href "" "database_builder_libs.models.abstract_store.AbstractStore"
            

TypeDB-backed implementation of the AbstractStore knowledge persistence layer.

This adapter maps canonical Node objects onto a TypeDB schema.

Mapping rules

Node → TypeDB Entity node.entity_type → entity type node.id → key_attribute value node.payload_data → attributes node.relations → relations

Identity semantics

A Node is uniquely identified by id.

Storing the same node twice MUST NOT create duplicates. The adapter performs existence checks before insertion.

Filter language

All read and delete operations use a URL-query style filter string:

"entity=<type>&<attribute>=<value>&<attribute>=<value>"

Examples entity=person&email=a@b.com entity=document&title=Report.pdf

Behaviour

• get_nodes() returns normalized Node objects reconstructed from TypeDB
• store_node() performs upsert semantics
• remove_node() deletes exactly one node (safety enforced)
• remove_nodes() allows bulk deletion when explicitly enabled

Safety guarantees

• Refuses full-entity deletion without explicit permission
• Deduplicates overlapping attribute matches
• Ensures relations are not duplicated
• Schema is automatically applied at initialization

Transactions

Each operation runs in its own transaction. Writes are committed automatically when successful.

Notes

This adapter assumes the schema defines a key attribute for each entity type.

Methods:

• connect –

  Establish connection to the backend.

• get_nodes –

  Retrieve nodes using a filter query.

• remove_node –

  Delete exactly one node and return it.

• remove_nodes –

  Delete nodes matching the filter.

• store_node –

  Insert a Node into TypeDB if it does not already exist, and ensure its relations exist.

Source code in src/database_builder_libs/stores/typedb/typedb_store.py

def __init__(self) -> None:
    super().__init__()
    self.typedb_driver: Driver | None = None
    self.database: str | None = None
    self._entity_attr_cache: dict[str, list[str]] = {}
    self._all_attr_cache: list[str] | None = None
    self._key_attr_cache: dict[str, str | None] = {}

connect

connect(config: dict | None = None) -> None

Establish connection to the backend.

This method is idempotent. Calling it multiple times must be safe.

Parameters

config : Any | None Backend-specific configuration object.

Raises

ConnectionError Backend unreachable. RuntimeError Backend misconfigured.

Source code in src/database_builder_libs/models/abstract_store.py

def connect(self, config: dict | None = None) -> None:
    """
    Establish connection to the backend.

    This method is idempotent. Calling it multiple times must be safe.

    Parameters
    ----------
    config : Any | None
        Backend-specific configuration object.

    Raises
    ------
    ConnectionError
        Backend unreachable.
    RuntimeError
        Backend misconfigured.
    """
    if self._connected:
        return

    self._connecting = True
    try:
        self._connect_impl(config)
        self._connected = True
    finally:
        self._connecting = False

get_nodes

get_nodes(filter: str | None) -> list[Node]

Retrieve nodes using a filter query.

Filter syntax

URL query string:

entity=<entity_type>&<attr>=<value>&...

Example entity=person&email=john@doe.com

Behaviour

• Returns normalized Node objects
• Deduplicates overlapping matches
• Automatically infers key_attribute from schema

Raises

TypeError If filter is not a string ValueError If filter missing required entity parameter

Source code in src/database_builder_libs/stores/typedb/typedb_store.py

def get_nodes(self, filter: str | None) -> list[Node]:
    """
    Retrieve nodes using a filter query.

    Filter syntax
    -------------
    URL query string:

        entity=<entity_type>&<attr>=<value>&...

    Example
        entity=person&email=john@doe.com

    Behaviour
    ---------
    - Returns normalized Node objects
    - Deduplicates overlapping matches
    - Automatically infers key_attribute from schema

    Raises
    ------
    TypeError
        If filter is not a string
    ValueError
        If filter missing required entity parameter
    """

    if filter is None:
        return self._get_all_nodes()

    if not isinstance(filter, str):
        raise TypeError("filter must be a string or None")

    if not filter:
        raise ValueError("filter cannot be empty; use None to fetch all nodes")

    self._ensure_connected()

    parsed = self._parse_filter(filter)
    entity_type = parsed["entity_type"]
    attrs = parsed["attributes"]
    include_relations = parsed["include_relations"]

    match_block = self._build_match(entity_type, attrs)

    attr_labels = self._get_entity_attribute_labels(entity_type)
    if not attr_labels:
        return []

    query = f"""
    match
        {match_block};
    fetch {{
        'data': {{$e.*}},
        'entity_type': '{entity_type}',
    }};
    """

    rows = self.query_read(query).as_concept_documents()

    return self._fetch_to_nodes(list(rows), include_relations=include_relations)

remove_node

remove_node(filter: str) -> Node

Delete exactly one node and return it.

Raises

ValueError No match or multiple matches RuntimeError Deletion inconsistency detected

Source code in src/database_builder_libs/stores/typedb/typedb_store.py

def remove_node(self, filter: str) -> Node:
    """
    Delete exactly one node and return it.

    Raises
    ------
    ValueError
        No match or multiple matches
    RuntimeError
        Deletion inconsistency detected
    """

    nodes = self.get_nodes(filter)

    if not nodes:
        raise ValueError(f"No node found for filter: {filter}")

    if len(nodes) > 1:
        raise ValueError(
            "remove_node() matched multiple nodes. "
            "Use remove_nodes(..., allow_multiple=True) instead."
        )

    removed = nodes[0]

    deleted_count = self.remove_nodes(filter, allow_multiple=False)

    if deleted_count != 1:
        raise RuntimeError("TypeDB deletion inconsistency detected")

    return removed

remove_nodes

remove_nodes(filter: str, allow_multiple: bool = False) -> int

Delete nodes matching the filter.

Safety rules

• Refuses deleting entire entity type unless allow_multiple=True
• Requires at least one attribute filter by default

Returns

int Number of deleted nodes

This operation is irreversible.

Source code in src/database_builder_libs/stores/typedb/typedb_store.py

def remove_nodes(self, filter: str, allow_multiple: bool = False) -> int:
    """
    Delete nodes matching the filter.

    Safety rules
    ------------
    - Refuses deleting entire entity type unless allow_multiple=True
    - Requires at least one attribute filter by default

    Returns
    -------
    int
        Number of deleted nodes

    This operation is irreversible.
    """

    if not filter:
        raise ValueError("TypeDB datastore requires a keyed filter string")

    parsed = self._parse_filter(filter)
    entity_type: str = parsed["entity_type"]
    attrs: dict[str, str] = parsed["attributes"]
    self._ensure_connected()

    if not allow_multiple and not attrs:
        raise ValueError(
            f"Refusing to delete all instances of '{entity_type}'. "
            "Provide a keyed filter or set allow_multiple=True."
        )

    match_block = self._build_match(entity_type, attrs)

    delete_query = f"""
    match
        {match_block};
    delete
        $e;
    """

    nodes = self.get_nodes(filter)
    count = len(nodes)

    if count == 0:
        return 0

    self.query_write(delete_query)

    return count

store_node

store_node(node: Node) -> None

Insert a Node into TypeDB if it does not already exist, and ensure its relations exist.

Behaviour

• Creates the entity if it does not exist
• Leaves existing entity attributes unchanged (no merge or update of payload_data)
• Inserts missing relations only
• Operation is idempotent

Identity rule

A node is considered existing if an entity with (entity_type, key_attribute, id) exists.

Relations are inserted only if not already present.

Source code in src/database_builder_libs/stores/typedb/typedb_store.py

def store_node(self, node: Node) -> None:
    """
    Insert a Node into TypeDB if it does not already exist, and ensure its relations exist.

    Behaviour
    ---------
    - Creates the entity if it does not exist
    - Leaves existing entity attributes unchanged (no merge or update of payload_data)
    - Inserts missing relations only
    - Operation is idempotent

    Identity rule
    -------------
    A node is considered existing if an entity with
    (entity_type, key_attribute, id) exists.

    Relations are inserted only if not already present.
    """
    self._ensure_connected()
    if not self._entity_exists(
        node.entity_type,
        node.key_attribute,
        node.id,
    ):
        self._insert_entity(
            node.entity_type,
            node.key_attribute,
            node.id,
            node.payload_data,
        )

    for rel in node.relations:
        self._insert_relation(cast(RelationData, rel))