`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)

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`

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,
):
    """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.
    """
    self.address = address
    self.database_name = database
    self.username = username
    self.password = password
    self.http_port = http_port
    self.server_version = server_version
    self._driver: Driver | None = driver
    self._owns_driver: bool = driver is None  # Track ownership

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, it will be closed.

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, it will be closed.
    """
    if self._driver:
        if self._owns_driver:
            logger.debug(f"Closing connection to TypeDB at {self.address}")
            self._driver.close()
            logger.info(f"Disconnected from TypeDB at {self.address}")
        else:
            logger.debug("Clearing driver reference (external driver, not closing)")
        self._driver = None
    if hasattr(self, "_rust_backend_database"):
        delattr(self, "_rust_backend_database")

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."""
    if self._driver is not None and self._owns_driver:
        warnings.warn(
            f"Database connection to {self.address} was not closed. "
            "Use 'with Database(...) as db:' or call db.close() explicitly.",
            ResourceWarning,
            stacklevel=2,
        )
        # Attempt to close to prevent resource leak
        try:
            self._driver.close()
        except Exception:
            pass  # Ignore errors during cleanup

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."""
    if self._driver is not None:
        if not self.driver.databases.contains(self.database_name):
            logger.debug(f"Creating database: {self.database_name}")
            self.driver.databases.create(self.database_name)
            logger.info(f"Database created: {self.database_name}")
        else:
            logger.debug(f"Database already exists: {self.database_name}")
        return

    from type_bridge._rust_runtime import rust_database_for

    rust_db = rust_database_for(self)
    if not rust_db.database_exists():
        logger.debug(f"Creating database: {self.database_name}")
        rust_db.create_database()
        logger.info(f"Database created: {self.database_name}")
    else:
        logger.debug(f"Database already exists: {self.database_name}")

delete_database ¶

delete_database()

Delete the database.

Source code in type_bridge/session.py

def delete_database(self) -> None:
    """Delete the database."""
    if self._driver is not None:
        if self.driver.databases.contains(self.database_name):
            logger.debug(f"Deleting database: {self.database_name}")
            self.driver.databases.get(self.database_name).delete()
            logger.info(f"Database deleted: {self.database_name}")
        else:
            logger.debug(f"Database does not exist, skipping delete: {self.database_name}")
        return

    from type_bridge._rust_runtime import rust_database_for

    rust_db = rust_database_for(self)
    if rust_db.database_exists():
        logger.debug(f"Deleting database: {self.database_name}")
        rust_db.delete_database()
        logger.info(f"Database deleted: {self.database_name}")
    else:
        logger.debug(f"Database does not exist, skipping delete: {self.database_name}")

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

transaction ¶

transaction(transaction_type: TransactionType) -> TransactionContext

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

transaction(transaction_type='read')

Create a transaction context.

Parameters:

Name	Type	Description	Default
`transaction_type`	`TransactionType \| 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: TransactionType | 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 = 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}")
    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.0") when known. None only when the connection was established through the band-7 gRPC fallback, where the server cannot report its version — supply server_version= at construction for strict validation there.

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.0"``) when known. ``None``
    only when the connection was established through the band-7 gRPC
    fallback, where the server cannot report its version — supply
    ``server_version=`` at construction for strict validation there.
    """
    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()

manager ¶

manager(model_cls)

Get a TypeDBManager bound to this transaction.

Source code in type_bridge/session.py

def manager(self, model_cls: Any):
    """Get a TypeDBManager bound to this transaction."""
    from type_bridge.models import Entity, Relation

    if issubclass(model_cls, (Entity, Relation)):
        return model_cls.manager(self)

    raise TypeError("manager() expects an Entity or Relation subclass")

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)

type_bridge.session¶

session ¶

Database ¶

driver property ¶

connect ¶

close ¶

__enter__ ¶

__exit__ ¶

__del__ ¶

create_database ¶

delete_database ¶

database_exists ¶

transaction ¶

execute_query ¶

detected_server_version ¶

check_schema_annotation_support ¶

supports_given_stage ¶

execute_with_rows ¶

get_schema ¶

Transaction ¶

is_open property ¶

execute ¶

commit ¶

rollback ¶

close ¶

TransactionContext ¶

transaction property ¶

database property ¶

execute ¶

execute_with_rows ¶

commit ¶

rollback ¶

manager ¶

ConnectionExecutor ¶

has_transaction property ¶

database property ¶

transaction property ¶

execute ¶

`type_bridge.session`¶

driver `property` ¶

enter ¶

exit ¶

del ¶

is_open `property` ¶

transaction `property` ¶

database `property` ¶

has_transaction `property` ¶

database `property` ¶

transaction `property` ¶