type_bridge.crud

crud

CRUD operations for TypeDB entities and relations.

EntityNotFoundError

Bases: NotFoundError

Raised when an entity does not exist in the database.

This exception is raised during delete or update operations when the target entity cannot be found using its @key attributes or matched attributes.

Example

try: manager.delete(nonexistent_entity) except EntityNotFoundError: print("Entity was already deleted or never existed")

HydrationError

HydrationError(model_type, raw_data, cause)

Bases: RuntimeError

Raised when hydrating database results into model instances fails.

This exception is raised when the ORM cannot convert raw database results into Python model instances. This typically indicates: - Schema mismatch between database and Python models - Corrupt or unexpected data in the database - Missing or invalid type information

Attributes:

Name	Type	Description
model_type		Name of the model class being hydrated
raw_data		The raw database result that failed to hydrate
cause		The underlying exception that caused the failure

Example

try: results = manager.all() except HydrationError as e: print(f"Failed to hydrate {e.model_type}: {e.cause}") print(f"Raw data: {e.raw_data}")

Source code in type_bridge/crud/exceptions.py

def __init__(
    self,
    model_type: str,
    raw_data: object,
    cause: Exception,
):
    self.model_type = model_type
    self.raw_data = raw_data
    self.cause = cause

    message = (
        f"Failed to hydrate {model_type} from database result: {cause}. "
        f"This may indicate a schema mismatch or corrupt data. "
        f"Raw data: {raw_data!r}"
    )
    super().__init__(message)

KeyAttributeError

KeyAttributeError(entity_type, operation, field_name=None, all_fields=None)

Bases: ValueError

Raised when @key attribute validation fails during update/delete.

This exception is raised when: - A @key attribute has a None value - No @key attributes are defined on the entity

Attributes:

Name	Type	Description
entity_type		Name of the entity class
operation		The operation that failed ("update" or "delete")
field_name		The @key field that was None (if applicable)
all_fields		List of all defined fields (when no @key exists)

Example

try: manager.update(entity_with_none_key) except KeyAttributeError as e: print(f"Key validation failed: {e}") print(f"Entity type: {e.entity_type}") print(f"Operation: {e.operation}")

Source code in type_bridge/crud/exceptions.py

def __init__(
    self,
    entity_type: str,
    operation: str,
    field_name: str | None = None,
    all_fields: list[str] | None = None,
):
    self.entity_type = entity_type
    self.operation = operation
    self.field_name = field_name
    self.all_fields = all_fields

    if field_name is not None:
        # Key attribute is None
        message = (
            f"Cannot {operation} {entity_type}: "
            f"key attribute '{field_name}' is None. "
            f"Ensure the entity has a valid '{field_name}' value "
            f"before calling {operation}()."
        )
    else:
        # No @key attributes defined
        message = (
            f"Cannot {operation} {entity_type}: no @key attributes found. "
            f"The {operation}() method requires at least one @key attribute "
            f"to identify the entity. "
            f"Defined attributes: {all_fields} (none marked as @key). "
            f"Hint: Add Flag(Key) to an attribute, e.g., `id: Id = Flag(Key)`"
        )

    super().__init__(message)

NotFoundError

Bases: LookupError

Base class for not-found errors in CRUD operations.

Raised when an entity or relation that was expected to exist cannot be found in the database.

NotUniqueError

Bases: ValueError

Raised when an operation requires exactly one match but finds multiple.

This exception is raised when attempting to delete an entity without @key attributes and multiple matching records are found. Use filter().delete() for bulk deletion instead.

Example

try: manager.delete(keyless_entity) except NotUniqueError: print("Multiple entities matched - use filter().delete() for bulk deletion")

RelationNotFoundError

Bases: NotFoundError

Raised when a relation does not exist in the database.

This exception is raised during delete or update operations when the target relation cannot be found using its role players' @key attributes.

Example

try: manager.delete(nonexistent_relation) except RelationNotFoundError: print("Relation was already deleted or never existed")

CrudEvent

Bases: Enum

CRUD lifecycle events.

CrudHook

Bases: Protocol

Protocol for CRUD lifecycle hooks.

Implement only the methods you need. All methods are optional — HookRunner uses hasattr / getattr to discover them.

HookCancelled

HookCancelled(reason='', *, event=None, hook=None)

Bases: Exception

Raise in a pre-hook to abort the operation.

Attributes:

Name	Type	Description
reason		Human-readable explanation.
event		The event that was cancelled (set by HookRunner).
hook		The hook instance that raised the cancellation (set by HookRunner).

Source code in type_bridge/crud/hooks.py

def __init__(
    self,
    reason: str = "",
    *,
    event: CrudEvent | None = None,
    hook: Any = None,
):
    self.reason = reason
    self.event = event
    self.hook = hook
    super().__init__(reason)

GroupByQuery

GroupByQuery(manager, filters, expressions, group_fields)

Rust-backed grouped aggregation query for exact-match Phase 3 filters.

Source code in type_bridge/crud/rust_manager.py

def __init__(
    self,
    manager: _RustQueryManager[T],
    filters: dict[str, Any],
    expressions: list[Any],
    group_fields: list[str],
):
    self._manager = manager
    self._filters = dict(filters)
    self._expressions = list(expressions)
    self._group_fields = group_fields

TypeDBManager

TypeDBManager(connection, model_class)

Manager that delegates CRUD and query execution to Rust.

Source code in type_bridge/crud/rust_manager.py

def __init__(self, connection: Connection, model_class: type[T]):
    self._connection = connection
    self.model_class = model_class
    self._descriptor = register_model_descriptor(model_class)
    self._manager_instance: Any | None = None
    if issubclass(model_class, Entity):
        self._kind = "entity"
    elif issubclass(model_class, Relation):
        self._kind = "relation"
    else:
        raise TypeError(f"Unsupported model type: {model_class}")
    self._hook_runner = HookRunner()

TypeDBQuery

TypeDBQuery(manager, filters, expressions=())

Rust-backed chainable query using the typed dynamic query surface.

Source code in type_bridge/crud/rust_manager.py

def __init__(
    self,
    manager: _RustQueryManager[T],
    filters: dict[str, Any],
    expressions: tuple[Any, ...] = (),
):
    self._manager = manager
    self._filters = dict(filters)
    self._expressions = list(expressions)
    self._order_fields: list[tuple[str, bool]] = []
    self._limit_value: int | None = None
    self._offset_value: int | None = None