type_bridge.session

session

Session and transaction management for TypeDB.

Database

Database(address='localhost:1729', database='typedb', username=None, password=None, driver=None, *, http_port=DEFAULT_HTTP_PORT, server_version=None, tls=None, tls_root_ca=None)

Main database connection and session manager.

Initialize database connection.

Parameters:

Name	Type	Description	Default
address	str	TypeDB server address	'localhost:1729'
database	str	Database name	'typedb'
username	str | None	Optional username for authentication	None
password	str | None	Optional password for authentication	None
driver	Driver | None	Optional pre-existing Driver instance to use. If provided, the Database will use this driver instead of creating a new one. The caller retains ownership and is responsible for closing it.	None
http_port	int	TypeDB HTTP API port used by the connect-time version gate probe (default 8000).	DEFAULT_HTTP_PORT
server_version	str | None	Exact TypeDB server version to use for connect-time validation instead of probing the HTTP API. Use this for gRPC-only deployments with the HTTP API disabled.	None
tls	bool | None	Explicit TLS policy. True uses native roots, False disables TLS, and omission preserves the released exact lowercase https:// address-prefix inference.	None
tls_root_ca	str | PathLike[str] | None	PEM root-CA path for an explicitly enabled TLS connection. A root path never enables TLS implicitly.	None

Source code in type_bridge/session.py

def __init__(
    self,
    address: str = "localhost:1729",
    database: str = "typedb",
    username: str | None = None,
    password: str | None = None,
    driver: Driver | None = None,
    *,
    http_port: int = typedb_driver.DEFAULT_HTTP_PORT,
    server_version: str | None = None,
    tls: bool | None = None,
    tls_root_ca: str | os.PathLike[str] | None = None,
):
    """Initialize database connection.

    Args:
        address: TypeDB server address
        database: Database name
        username: Optional username for authentication
        password: Optional password for authentication
        driver: Optional pre-existing Driver instance to use. If provided,
            the Database will use this driver instead of creating a new one.
            The caller retains ownership and is responsible for closing it.
        http_port: TypeDB HTTP API port used by the connect-time version
            gate probe (default 8000).
        server_version: Exact TypeDB server version to use for connect-time
            validation instead of probing the HTTP API. Use this for
            gRPC-only deployments with the HTTP API disabled.
        tls: Explicit TLS policy. ``True`` uses native roots, ``False``
            disables TLS, and omission preserves the released exact
            lowercase ``https://`` address-prefix inference.
        tls_root_ca: PEM root-CA path for an explicitly enabled TLS
            connection. A root path never enables TLS implicitly.
    """
    # Establish destructor-safe ownership state before any validation can
    # raise. Python may invoke ``__del__`` for an object whose ``__init__``
    # exited early, including the fail-closed TLS checks below.
    self._driver: Driver | None = driver
    self._owns_driver: bool = driver is None
    self._tls_root_ca_snapshot: Any | None = None
    self._tls_root_ca_snapshot_path: str | None = None
    self._transport_lock = threading.RLock()
    self._prepared_connection: _PreparedConnection | None = None
    self._transport_committed = False
    self.address = address

    # Reject contradictory or ill-typed policy before any connect path can
    # create a native host. The resolved value is recalculated at connect
    # time so mutation of the released public attributes keeps working.
    _, _, normalized_tls_root_ca = _resolve_transport_options(address, tls, tls_root_ca)
    self.database_name = database
    self.username = username
    self.password = password
    self.http_port = http_port
    self.server_version = server_version
    self.tls = tls
    self.tls_root_ca = normalized_tls_root_ca

driver property

driver

Get the TypeDB driver, connecting if necessary.

connect

connect()

Connect to TypeDB server through the Rust runtime.

If a driver was injected via init, this method does nothing (the driver is already connected). Otherwise, initializes the cached Rust database handle. Direct access to the external Python TypeDB driver remains available through the driver property.

Source code in type_bridge/session.py

def connect(self) -> None:
    """Connect to TypeDB server through the Rust runtime.

    If a driver was injected via __init__, this method does nothing
    (the driver is already connected). Otherwise, initializes the cached
    Rust database handle. Direct access to the external Python TypeDB
    driver remains available through the ``driver`` property.
    """
    if self._driver is not None:
        return

    logger.debug(f"Connecting to TypeDB at {self.address} (database: {self.database_name})")
    from type_bridge._backend import selected_backend
    from type_bridge._rust_runtime import rust_database_for

    selected_backend()
    rust_database_for(self)
    logger.info(f"Connected to TypeDB at {self.address}")

close

close()

Close connection to TypeDB server.

If the driver was injected via init, this method only clears the reference without closing the driver (the caller retains ownership). If the driver was created internally, the owned Python driver closes first. If that close fails, the complete transport remains attached for a released-style retry. After it succeeds, embedded-Rust and snapshot cleanup are attempted. A Rust close failure is logged and masked to preserve the released Python Database.close() contract; snapshot failures retain their normal error behavior.

Source code in type_bridge/session.py

def close(self) -> None:
    """Close connection to TypeDB server.

    If the driver was injected via __init__, this method only clears the
    reference without closing the driver (the caller retains ownership).
    If the driver was created internally, the owned Python driver closes
    first. If that close fails, the complete
    transport remains attached for a released-style retry. After it
    succeeds, embedded-Rust and snapshot cleanup are attempted. A Rust
    close failure is logged and masked to preserve the released Python
    ``Database.close()`` contract; snapshot failures retain their normal
    error behavior.
    """
    first_error: Exception | None = None
    with self._transport_lock:
        # Detach every owned resource before calling external cleanup code.
        # This makes repeated and re-entrant close calls harmless even when
        # one of the cleanup operations fails.
        driver = self._driver
        owns_driver = self._owns_driver
        # The released close path used `if self._driver:` rather than an
        # identity check. Preserve that observable injection seam: a
        # falsey driver double is neither closed nor detached, and an
        # exception from its truth probe occurs before any cleanup state
        # is mutated.
        released_driver_present = bool(driver) if driver is not None else False
        had_rust_database = hasattr(self, "_rust_backend_database")
        rust_database = getattr(self, "_rust_backend_database", None)
        snapshot = self._tls_root_ca_snapshot
        snapshot_path = self._tls_root_ca_snapshot_path
        prepared = self._prepared_connection
        transport_committed = self._transport_committed
        if released_driver_present:
            self._driver = None
        if hasattr(self, "_rust_backend_database"):
            delattr(self, "_rust_backend_database")
        self._tls_root_ca_snapshot = None
        self._tls_root_ca_snapshot_path = None
        self._prepared_connection = None
        self._transport_committed = False

        python_close_failed = False
        if released_driver_present:
            assert driver is not None
            if owns_driver:
                logger.debug(f"Closing connection to TypeDB at {self.address}")
                try:
                    driver.close()
                except Exception as error:
                    first_error = error
                    python_close_failed = True
                else:
                    logger.info(f"Disconnected from TypeDB at {self.address}")
            else:
                logger.debug("Clearing driver reference (external driver, not closing)")

        restored_failed_transport = False
        if python_close_failed and self._driver is None:
            # Released close() left the complete owned transport attached
            # when its first (Python-driver) close raised. Restore that
            # ordering and retry state unless re-entrant work installed a
            # replacement while the external close callback was running.
            self._driver = driver
            self._owns_driver = owns_driver
            if had_rust_database:
                setattr(self, "_rust_backend_database", rust_database)
            self._tls_root_ca_snapshot = snapshot
            self._tls_root_ca_snapshot_path = snapshot_path
            self._prepared_connection = prepared
            self._transport_committed = transport_committed
            restored_failed_transport = True

        if not restored_failed_transport and rust_database is not None:
            try:
                rust_close = getattr(rust_database, "close", None)
                # Older test doubles and third-party compatibility shims
                # may predate the explicit native-close seam. Real Rust
                # handles always expose it; V1 stand-ins remain safely
                # releasable.
                if callable(rust_close):
                    rust_close()
            except Exception:
                logger.warning("Embedded Rust backend cleanup failed; releasing the handle")

        if not restored_failed_transport and snapshot is not None:
            try:
                snapshot.cleanup()
            except Exception as error:
                if first_error is None:
                    first_error = error

    if first_error is not None:
        raise first_error

__getstate__

__getstate__()

Preserve released pickling for pristine connection configs.

Source code in type_bridge/session.py

def __getstate__(self) -> dict[str, Any]:
    """Preserve released pickling for pristine connection configs."""
    with self._transport_lock:
        if self._prepared_connection is not None:
            raise TypeError("a Database with prepared transport cannot be pickled")
        state = self.__dict__.copy()
        state.pop("_transport_lock", None)
        return state

__enter__

__enter__()

Context manager entry.

Source code in type_bridge/session.py

def __enter__(self) -> Database:
    """Context manager entry."""
    self.connect()
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Source code in type_bridge/session.py

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    """Context manager exit."""
    del exc_type, exc_val, exc_tb  # unused
    self.close()

__del__

__del__()

Destructor that warns if driver was not properly closed.

Source code in type_bridge/session.py

def __del__(self) -> None:
    """Destructor that warns if driver was not properly closed."""
    lock = getattr(self, "_transport_lock", None)
    if lock is None:
        return
    with lock:
        driver = getattr(self, "_driver", None)
        if driver is not None and getattr(self, "_owns_driver", False):
            warnings.warn(
                f"Database connection to {getattr(self, 'address', '<unknown>')} was not "
                "closed. Use 'with Database(...) as db:' or call db.close() explicitly.",
                ResourceWarning,
                stacklevel=2,
            )
            # Attempt to close to prevent resource leak
            try:
                driver.close()
            except Exception:
                pass  # Ignore errors during cleanup
        snapshot = getattr(self, "_tls_root_ca_snapshot", None)
        if snapshot is not None:
            try:
                snapshot.cleanup()
            except Exception:
                pass

create_database

create_database()

Create the database if it doesn't exist.

Source code in type_bridge/session.py

def create_database(self) -> None:
    """Create the database if it doesn't exist."""
    self.create_database_outcome()

create_database_outcome

create_database_outcome()

Create the bound database and return a race-normalized outcome.

Source code in type_bridge/session.py

def create_database_outcome(self) -> Literal["created", "already_exists"]:
    """Create the bound database and return a race-normalized outcome."""
    if self._driver is not None:
        if self.driver.databases.contains(self.database_name):
            logger.debug(f"Database already exists: {self.database_name}")
            return "already_exists"
        logger.debug(f"Creating database: {self.database_name}")
        try:
            self.driver.databases.create(self.database_name)
        except Exception:
            if self.driver.databases.contains(self.database_name):
                return "already_exists"
            raise
        logger.info(f"Database created: {self.database_name}")
        return "created"

    from type_bridge._rust_runtime import rust_database_for

    outcome = cast(
        Literal["created", "already_exists"],
        rust_database_for(self).create_database_outcome(),
    )
    logger.debug(f"Database create outcome for '{self.database_name}': {outcome}")
    return outcome

delete_database

delete_database()

Delete the database.

Source code in type_bridge/session.py

def delete_database(self) -> None:
    """Delete the database."""
    self.delete_database_outcome()

delete_database_outcome

delete_database_outcome()

Delete the bound database and return a race-normalized outcome.

Source code in type_bridge/session.py

def delete_database_outcome(self) -> Literal["deleted", "already_absent"]:
    """Delete the bound database and return a race-normalized outcome."""
    if self._driver is not None:
        if not self.driver.databases.contains(self.database_name):
            logger.debug(f"Database does not exist, skipping delete: {self.database_name}")
            return "already_absent"
        logger.debug(f"Deleting database: {self.database_name}")
        try:
            self.driver.databases.get(self.database_name).delete()
        except Exception:
            if not self.driver.databases.contains(self.database_name):
                return "deleted"
            raise
        logger.info(f"Database deleted: {self.database_name}")
        return "deleted"

    from type_bridge._rust_runtime import rust_database_for

    outcome = cast(
        Literal["deleted", "already_absent"],
        rust_database_for(self).delete_database_outcome(),
    )
    logger.debug(f"Database delete outcome for '{self.database_name}': {outcome}")
    return outcome

database_exists

database_exists()

Check if database exists.

Source code in type_bridge/session.py

def database_exists(self) -> bool:
    """Check if database exists."""
    if self._driver is not None:
        exists = self.driver.databases.contains(self.database_name)
    else:
        from type_bridge._rust_runtime import rust_database_for

        exists = rust_database_for(self).database_exists()
    logger.debug(f"Database exists check for '{self.database_name}': {exists}")
    return exists

inspect_database_pair

inspect_database_pair()

Inspect the managed database and its package-owned journal as one pair.

Source code in type_bridge/session.py

def inspect_database_pair(
    self,
) -> Literal["absent", "standalone_managed", "owned_pair", "owned_journal_orphan"]:
    """Inspect the managed database and its package-owned journal as one pair."""
    from type_bridge._rust_runtime import rust_database_for

    return cast(
        Literal["absent", "standalone_managed", "owned_pair", "owned_journal_orphan"],
        rust_database_for(self).inspect_database_pair(),
    )

inspect_database_pair_controlled

inspect_database_pair_controlled(*, timeout_milliseconds=None, cancellation=None)

Inspect the managed database pair under explicit execution controls.

Source code in type_bridge/session.py

def inspect_database_pair_controlled(
    self,
    *,
    timeout_milliseconds: int | None = None,
    cancellation: Any | None = None,
) -> Literal["absent", "standalone_managed", "owned_pair", "owned_journal_orphan"]:
    """Inspect the managed database pair under explicit execution controls."""
    from type_bridge._rust_runtime import rust_database_for

    return cast(
        Literal["absent", "standalone_managed", "owned_pair", "owned_journal_orphan"],
        rust_database_for(self).inspect_database_pair_controlled(
            timeout_milliseconds=timeout_milliseconds,
            cancellation=cancellation,
        ),
    )

plan_database_delete

plan_database_delete()

Create a single-use, pair-aware managed database deletion plan.

Source code in type_bridge/session.py

def plan_database_delete(self) -> Any:
    """Create a single-use, pair-aware managed database deletion plan."""
    from type_bridge._rust_runtime import rust_database_for

    return rust_database_for(self).plan_database_delete()

plan_database_delete_controlled

plan_database_delete_controlled(*, timeout_milliseconds=None, cancellation=None)

Create a managed deletion plan under explicit execution controls.

Source code in type_bridge/session.py

def plan_database_delete_controlled(
    self,
    *,
    timeout_milliseconds: int | None = None,
    cancellation: Any | None = None,
) -> Any:
    """Create a managed deletion plan under explicit execution controls."""
    from type_bridge._rust_runtime import rust_database_for

    return rust_database_for(self).plan_database_delete_controlled(
        timeout_milliseconds=timeout_milliseconds,
        cancellation=cancellation,
    )

transaction

transaction(transaction_type: Enum) -> TransactionContext

transaction(transaction_type: str = 'read') -> TransactionContext

transaction(transaction_type='read')

Create a transaction context.

Parameters:

Name	Type	Description	Default
transaction_type	Enum | str	TransactionType or string ("read", "write", "schema")	'read'

Returns:

Type	Description
TransactionContext	TransactionContext for use as a context manager

Source code in type_bridge/session.py

def transaction(self, transaction_type: Enum | str = "read") -> TransactionContext:
    """Create a transaction context.

    Args:
        transaction_type: TransactionType or string ("read", "write", "schema")

    Returns:
        TransactionContext for use as a context manager
    """
    tx_type_map: dict[str, TransactionType] = {
        "read": TransactionType.READ,
        "write": TransactionType.WRITE,
        "schema": TransactionType.SCHEMA,
    }

    if isinstance(transaction_type, str):
        tx_type = tx_type_map.get(transaction_type, TransactionType.READ)
    else:
        tx_type = cast(TransactionType, transaction_type)

    logger.debug(
        f"Creating {_tx_type_name(tx_type)} transaction for database: {self.database_name}"
    )
    return TransactionContext(self, tx_type)

execute_query

execute_query(query, transaction_type='read')

Execute a query and return results.

Parameters:

Name	Type	Description	Default
query	str	TypeQL query string	required
transaction_type	str	Type of transaction ("read", "write", or "schema")	'read'

Returns:

Type	Description
list[dict[str, Any]]	List of result dictionaries

Source code in type_bridge/session.py

def execute_query(self, query: str, transaction_type: str = "read") -> list[dict[str, Any]]:
    """Execute a query and return results.

    Args:
        query: TypeQL query string
        transaction_type: Type of transaction ("read", "write", or "schema")

    Returns:
        List of result dictionaries
    """
    logger.debug(f"Executing query (type={transaction_type}, {len(query)} chars)")
    logger.debug(f"Query: {query}")
    if transaction_type in ("schema", TransactionType.SCHEMA):
        self.check_schema_annotation_support(query)
    with self.transaction(transaction_type) as tx:
        results = tx.execute(query)
        if isinstance(transaction_type, str):
            needs_commit = transaction_type in ("write", "schema")
        else:
            needs_commit = transaction_type in (TransactionType.WRITE, TransactionType.SCHEMA)
        if needs_commit:
            tx.commit()
        logger.debug(f"Query returned {len(results)} results")
        return results

detected_server_version

detected_server_version()

The server version detected by the connect-time version gate.

Returns the version string (e.g. "3.12.3") when known. None means the negotiated connection path produced no authoritative server identity; supply server_version= at construction when strict identity validation is required.

Source code in type_bridge/session.py

def detected_server_version(self) -> str | None:
    """The server version detected by the connect-time version gate.

    Returns the version string (e.g. ``"3.12.3"``) when known. ``None``
    means the negotiated connection path produced no authoritative server
    identity; supply ``server_version=`` at construction when strict
    identity validation is required.
    """
    from type_bridge._rust_runtime import rust_database_for

    return rust_database_for(self).server_version()

check_schema_annotation_support

check_schema_annotation_support(typeql)

Version-gate schema DDL that uses @doc/@meta annotations.

Raises the versioned error when the TypeQL uses schema annotations (TypeDB 3.12+) and the detected server version predates 3.12. When the server version is unknown, the DDL is sent as-is and the server decides.

Source code in type_bridge/session.py

def check_schema_annotation_support(self, typeql: str) -> None:
    """Version-gate schema DDL that uses ``@doc``/``@meta`` annotations.

    Raises the versioned error when the TypeQL uses schema annotations
    (TypeDB 3.12+) and the detected server version predates 3.12. When
    the server version is unknown, the DDL is sent as-is and the server
    decides.
    """
    from type_bridge._rust_runtime import rust_database_for

    rust_database_for(self).check_schema_annotation_support(typeql)

supports_given_stage

supports_given_stage()

Whether given rows can execute on the active connection.

This requires both TypeDB 3.12+ syntax support and a negotiated band-9 provider. It remains False when the server version is unknown or when a 3.12 server stays on the safe band-8 discovery connection after a band-9 upgrade failure. Bulk operations consult this before dispatch and use their per-row fallback when it is False.

Source code in type_bridge/session.py

def supports_given_stage(self) -> bool:
    """Whether ``given`` rows can execute on the active connection.

    This requires both TypeDB 3.12+ syntax support and a negotiated band-9
    provider. It remains ``False`` when the server version is unknown or
    when a 3.12 server stays on the safe band-8 discovery connection after
    a band-9 upgrade failure. Bulk operations consult this before dispatch
    and use their per-row fallback when it is ``False``.
    """
    from type_bridge._rust_runtime import rust_database_for

    return rust_database_for(self).supports_given_stage()

execute_with_rows

execute_with_rows(query, transaction_type, variables, column_types, rows)

Execute a given-stage TypeQL query over input rows.

One compiled pipeline runs over every input row; the rows travel through the driver API instead of being interpolated into the query string, so user-supplied values never touch TypeQL text. Requires a TypeDB 3.12+ server; on older servers this raises the versioned error from the feature gate.

Parameters:

Name	Type	Description	Default
query	str	TypeQL starting with a given stage, e.g. given $n: string; insert $p isa person, has name == $n;	required
transaction_type	str	"read", "write", or "schema"	required
variables	list[str]	given variable names without the $ sigil, in column order	required
column_types	list[str]	TypeQL value type names aligned with variables ("string", "integer", "double", "boolean", "date", "datetime", "datetime-tz")	required
rows	list[list[Any]]	input rows, each a list of primitives in column order (temporal values as ISO-8601 strings)	required

Returns:

Type	Description
list[dict[str, Any]]	List of result dictionaries (one per pipeline output row).

Source code in type_bridge/session.py

def execute_with_rows(
    self,
    query: str,
    transaction_type: str,
    variables: list[str],
    column_types: list[str],
    rows: list[list[Any]],
) -> list[dict[str, Any]]:
    """Execute a ``given``-stage TypeQL query over input rows.

    One compiled pipeline runs over every input row; the rows travel
    through the driver API instead of being interpolated into the query
    string, so user-supplied values never touch TypeQL text. Requires a
    TypeDB 3.12+ server; on older servers this raises the versioned
    error from the feature gate.

    Args:
        query: TypeQL starting with a ``given`` stage, e.g.
            ``given $n: string; insert $p isa person, has name == $n;``
        transaction_type: "read", "write", or "schema"
        variables: given variable names without the ``$`` sigil,
            in column order
        column_types: TypeQL value type names aligned with ``variables``
            ("string", "integer", "double", "boolean", "date",
            "datetime", "datetime-tz")
        rows: input rows, each a list of primitives in column order
            (temporal values as ISO-8601 strings)

    Returns:
        List of result dictionaries (one per pipeline output row).
    """
    from type_bridge._rust_runtime import rust_database_for

    return rust_database_for(self).execute_with_rows(
        query, transaction_type, variables, column_types, rows
    )

get_schema

get_schema()

Get the schema definition for this database.

Source code in type_bridge/session.py

def get_schema(self) -> str:
    """Get the schema definition for this database."""
    logger.debug(f"Fetching schema for database: {self.database_name}")
    if self._driver is not None:
        db = self.driver.databases.get(self.database_name)
        schema = db.schema()
    else:
        from type_bridge._rust_runtime import schema_text

        schema = schema_text(self)
    logger.debug(f"Schema fetched ({len(schema)} chars)")
    return schema

Transaction

Transaction(tx)

Wrapper around TypeDB transaction.

Initialize transaction wrapper.

Parameters:

Name	Type	Description	Default
tx	Transaction	TypeDB transaction	required

Source code in type_bridge/session.py

def __init__(self, tx: TypeDBTransaction):
    """Initialize transaction wrapper.

    Args:
        tx: TypeDB transaction
    """
    self._tx = tx

is_open property

is_open

Check if transaction is open.

execute

execute(query)

Execute a query.

Parameters:

Name	Type	Description	Default
query	str	TypeQL query string	required

Returns:

Type	Description
list[dict[str, Any]]	List of result dictionaries

Source code in type_bridge/session.py

def execute(self, query: str) -> list[dict[str, Any]]:
    """Execute a query.

    Args:
        query: TypeQL query string

    Returns:
        List of result dictionaries
    """
    logger.debug(f"Transaction.execute: query ({len(query)} chars)")
    logger.debug(f"Query: {query}")
    # Execute query - returns a Promise[QueryAnswer]
    promise = self._tx.query(query)
    answer = promise.resolve()

    # Process based on answer type
    results = []

    # Check if the answer has an iterator (for fetch/get queries)
    if hasattr(answer, "__iter__"):
        for item in answer:
            if hasattr(item, "as_dict"):
                # ConceptRow with as_dict method - extract values from concepts
                raw_dict = dict(item.as_dict())
                results.append(_extract_values_from_dict(raw_dict))
            elif hasattr(item, "as_json"):
                # Document with as_json method
                results.append(item.as_json())
            elif hasattr(item, "column_names") and hasattr(item, "get"):
                # ConceptRow - extract IID and concept info
                result = _extract_concept_row(item)
                results.append(result)
            else:
                # Try to convert to dict
                results.append(
                    dict(item) if hasattr(item, "__iter__") else {"result": str(item)}
                )

    logger.debug(f"Query executed, {len(results)} results returned")
    return results

commit

commit()

Commit the transaction.

Source code in type_bridge/session.py

def commit(self) -> None:
    """Commit the transaction."""
    logger.debug("Committing transaction")
    self._tx.commit()
    logger.info("Transaction committed")

rollback

rollback()

Rollback the transaction.

Source code in type_bridge/session.py

def rollback(self) -> None:
    """Rollback the transaction."""
    logger.debug("Rolling back transaction")
    self._tx.rollback()
    logger.info("Transaction rolled back")

close

close()

Close the transaction if open.

Source code in type_bridge/session.py

def close(self) -> None:
    """Close the transaction if open."""
    if self._tx.is_open():
        logger.debug("Closing transaction")
        self._tx.close()

TransactionContext

TransactionContext(db, tx_type)

Context manager for sharing a TypeDB transaction across operations.

Source code in type_bridge/session.py

def __init__(self, db: Database, tx_type: TransactionType):
    self.db = db
    self.tx_type = tx_type
    self._tx: Transaction | None = None
    self._rust_tx: Any | None = None
    self._rust_finalized = False

transaction property

transaction

Underlying transaction wrapper.

database property

database

Database backing this transaction.

execute

execute(query)

Execute a query within the active transaction.

Source code in type_bridge/session.py

def execute(self, query: str) -> list[dict[str, Any]]:
    """Execute a query within the active transaction."""
    if self._rust_tx is not None:
        return self._rust_tx.execute(query)
    return self.transaction.execute(query)

execute_with_rows

execute_with_rows(query, variables, column_types, rows)

Execute a given-stage query with input rows in this transaction.

See :meth:Database.execute_with_rows for the argument contract. Requires the Rust backend on a TypeDB 3.12+ connection.

Source code in type_bridge/session.py

def execute_with_rows(
    self,
    query: str,
    variables: list[str],
    column_types: list[str],
    rows: list[list[Any]],
) -> list[dict[str, Any]]:
    """Execute a ``given``-stage query with input rows in this transaction.

    See :meth:`Database.execute_with_rows` for the argument contract.
    Requires the Rust backend on a TypeDB 3.12+ connection.
    """
    if self._rust_tx is None:
        raise RuntimeError("execute_with_rows requires an open Rust-backend transaction")
    return self._rust_tx.execute_with_rows(query, variables, column_types, rows)

commit

commit()

Commit the active transaction.

Source code in type_bridge/session.py

def commit(self) -> None:
    """Commit the active transaction."""
    if self._rust_tx is not None:
        self._rust_tx.commit()
        self._rust_finalized = True
        return
    self.transaction.commit()

rollback

rollback()

Rollback the active transaction.

Source code in type_bridge/session.py

def rollback(self) -> None:
    """Rollback the active transaction."""
    if self._rust_tx is not None:
        self._rust_tx.rollback()
        self._rust_finalized = True
        return
    self.transaction.rollback()

ConnectionExecutor

ConnectionExecutor(connection)

Delegate that handles query execution across connection types.

This class encapsulates the logic for executing queries against different connection types (Database, Transaction, TransactionContext, or proxy equivalents), providing a unified interface for CRUD operations.

Initialize the executor with a connection.

Parameters:

Name	Type	Description	Default
connection	Connection	Database, Transaction, TransactionContext, or proxy equivalent	required

Source code in type_bridge/session.py

def __init__(self, connection: Connection):
    """Initialize the executor with a connection.

    Args:
        connection: Database, Transaction, TransactionContext, or proxy equivalent
    """
    if isinstance(connection, (TransactionContext, ProxyTransactionContext)):
        logger.debug("ConnectionExecutor initialized with TransactionContext")
        self._transaction: Transaction | ProxyTransaction | _RustTransactionView | None = (
            connection.transaction
        )
        self._database: Database | ProxyDatabase | None = None
    elif isinstance(connection, (Transaction, ProxyTransaction)):
        logger.debug("ConnectionExecutor initialized with Transaction")
        self._transaction = connection
        self._database = None
    else:
        logger.debug("ConnectionExecutor initialized with Database")
        self._transaction = None
        self._database = connection

has_transaction property

has_transaction

Check if using an existing transaction.

database property

database

Get database if available (for creating new transactions).

transaction property

transaction

Get transaction if available.

execute

execute(query, tx_type)

Execute query, using existing transaction or creating a new one.

Parameters:

Name	Type	Description	Default
query	str	TypeQL query string	required
tx_type	TransactionType	Transaction type (used only when creating new transaction)	required

Returns:

Type	Description
list[dict[str, Any]]	List of result dictionaries

Source code in type_bridge/session.py

def execute(self, query: str, tx_type: TransactionType) -> list[dict[str, Any]]:
    """Execute query, using existing transaction or creating a new one.

    Args:
        query: TypeQL query string
        tx_type: Transaction type (used only when creating new transaction)

    Returns:
        List of result dictionaries
    """
    if self._transaction:
        logger.debug("ConnectionExecutor: using existing transaction")
        return self._transaction.execute(query)
    assert self._database is not None
    logger.debug(f"ConnectionExecutor: creating new {_tx_type_name(tx_type)} transaction")
    with self._database.transaction(tx_type) as tx:
        return tx.execute(query)