Skip to content

type_bridge

type_bridge

Python SDK for generated TypeBridge applications and retained query APIs.

ProxyDatabase 

ProxyDatabase(proxy_url='http://localhost:8080', database='typedb', timeout=30)

Drop-in replacement for Database that routes queries through a type-bridge proxy server.

Instead of connecting directly to TypeDB, all queries are sent as HTTP requests to the proxy server's REST API. The proxy handles validation, interceptors (audit log, etc.), and forwarding to TypeDB.

Source code in type_bridge/proxy.py 
130
131
132
133
134
135
136
137
138
139
def __init__(
    self,
    proxy_url: str = "http://localhost:8080",
    database: str = "typedb",
    timeout: int = 30,
):
    self.proxy_url = proxy_url.rstrip("/")
    self.database_name = database
    self.timeout = timeout
    self._connected = False

connect 

connect()

Verify the proxy server is reachable via health check.

Source code in type_bridge/proxy.py 
141
142
143
144
145
146
147
148
149
150
151
152
def connect(self) -> None:
    """Verify the proxy server is reachable via health check."""
    try:
        health = self._http_get("/health")
        self._connected = True
        logger.info(
            "Connected to proxy at %s (version: %s)",
            self.proxy_url,
            health.get("version", "unknown"),
        )
    except Exception as e:
        raise ConnectionError(f"Failed to connect to proxy at {self.proxy_url}: {e}") from e

close 

close()

Close the proxy connection (clears connected state).

Source code in type_bridge/proxy.py 
154
155
156
157
def close(self) -> None:
    """Close the proxy connection (clears connected state)."""
    self._connected = False
    logger.debug("Proxy connection closed: %s", self.proxy_url)

transaction 

transaction(transaction_type='read')

Create a proxy transaction context.

Parameters:

Name Type Description Default
transaction_type Any

Transaction type string ("read", "write", "schema") or TransactionType enum value.

 'read'
Source code in type_bridge/proxy.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def transaction(self, transaction_type: Any = "read") -> ProxyTransactionContext:
    """Create a proxy transaction context.

    Args:
        transaction_type: Transaction type string ("read", "write", "schema")
            or TransactionType enum value.
    """
    if isinstance(transaction_type, str):
        tx_type = transaction_type
    else:
        # Handle TransactionType enum from typedb.driver
        name = getattr(transaction_type, "name", str(transaction_type))
        tx_type = name.lower()
    return ProxyTransactionContext(self, tx_type)

execute_query 

execute_query(query, transaction_type='read')

Execute a query through the proxy and return results.

Source code in type_bridge/proxy.py 
182
183
184
185
186
def execute_query(self, query: str, transaction_type: str = "read") -> list[dict[str, Any]]:
    """Execute a query through the proxy and return results."""
    logger.debug("Executing query via proxy (type=%s, %d chars)", transaction_type, len(query))
    results = self._send_raw_query(query, transaction_type)
    return results if isinstance(results, list) else [results]

get_schema 

get_schema()

Fetch the loaded schema from the proxy server.

Source code in type_bridge/proxy.py 
188
189
190
191
def get_schema(self) -> str:
    """Fetch the loaded schema from the proxy server."""
    resp = self._http_get("/schema")
    return json.dumps(resp) if isinstance(resp, dict) else str(resp)

ProxyError 

ProxyError(message, code='UNKNOWN', details=None)

Bases: Exception

Error returned by the proxy server.

Source code in type_bridge/proxy.py 
31
32
33
34
def __init__(self, message: str, code: str = "UNKNOWN", details: Any = None):
    super().__init__(message)
    self.code = code
    self.details = details

Query 

Query()

Builder for TypeQL queries.

Initialize query builder.

Source code in type_bridge/query/__init__.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self):
    """Initialize query builder."""
    self.match_clause = MatchClause(patterns=[])
    self.delete_clause = DeleteClause(statements=[])
    self.insert_clause = InsertClause(statements=[])
    self.fetch_clause = FetchClause(items=[])

    # Modifiers
    self.sort_clauses: list[tuple[str, str]] = []
    self.offset_val: int | None = None
    self.limit_val: int | None = None

    self.compiler = QueryCompiler()

match 

match(pattern)

Add a match clause.

Parameters:

Name Type Description Default
pattern str

TypeQL match pattern

 required

Returns:

Type Description
Query

Self for chaining
Source code in type_bridge/query/__init__.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def match(self, pattern: str) -> Query:
    """Add a match clause.

    Args:
        pattern: TypeQL match pattern

    Returns:
        Self for chaining
    """
    if pattern:
        # Clean string pattern
        clean_pattern = pattern.strip().rstrip(";")
        # RawPattern doesn't strictly require 'variable' if handled by compiler as raw string
        # But we made it subclass Pattern again with 'variable' moved to subclasses.
        # RawPattern definition: content: str
        self.match_clause.patterns.append(RawPattern(content=clean_pattern))
    return self

fetch 

fetch(variable, *attributes)

Add variables and attributes to fetch.

In TypeQL 3.x, fetch uses the syntax: fetch { $e.* } (fetch all attributes)

Parameters:

Name Type Description Default
variable str

Variable name to fetch (e.g., "$e")

 required
attributes str

Not used in TypeQL 3.x (kept for API compatibility)

 ()

Returns:

Type Description
Query

Self for chaining
Example

query.fetch("$e") # Fetches all attributes

Source code in type_bridge/query/__init__.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def fetch(self, variable: str, *attributes: str) -> Query:
    """Add variables and attributes to fetch.

    In TypeQL 3.x, fetch uses the syntax:
    fetch { $e.* }  (fetch all attributes)

    Args:
        variable: Variable name to fetch (e.g., "$e")
        attributes: Not used in TypeQL 3.x (kept for API compatibility)

    Returns:
        Self for chaining

    Example:
        query.fetch("$e")  # Fetches all attributes
    """
    # For TypeQL 3.x, default to wildcard fetch
    # Use variable name (without $) as the key
    key = variable.lstrip("$")
    self.fetch_clause.items.append(FetchWildcard(key=key, var=variable))
    return self

delete 

delete(pattern)

Add a delete clause.

Parameters:

Name Type Description Default
pattern str

TypeQL delete pattern

 required

Returns:

Type Description
Query

Self for chaining
Source code in type_bridge/query/__init__.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
def delete(self, pattern: str) -> Query:
    """Add a delete clause.

    Args:
        pattern: TypeQL delete pattern

    Returns:
        Self for chaining
    """
    if pattern:
        clean_pattern = pattern.strip().rstrip(";")
        self.delete_clause.statements.append(RawStatement(content=clean_pattern))
    return self

insert 

insert(pattern)

Add an insert clause.

Parameters:

Name Type Description Default
pattern str

TypeQL insert pattern

 required

Returns:

Type Description
Query

Self for chaining
Source code in type_bridge/query/__init__.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def insert(self, pattern: str) -> Query:
    """Add an insert clause.

    Args:
        pattern: TypeQL insert pattern

    Returns:
        Self for chaining
    """
    if pattern:
        clean_pattern = pattern.strip().rstrip(";")
        self.insert_clause.statements.append(RawStatement(content=clean_pattern))
    return self

limit 

limit(limit)

Set query limit.

Parameters:

Name Type Description Default
limit int

Maximum number of results

 required

Returns:

Type Description
Query

Self for chaining
Source code in type_bridge/query/__init__.py 
114
115
116
117
118
119
120
121
122
123
124
def limit(self, limit: int) -> Query:
    """Set query limit.

    Args:
        limit: Maximum number of results

    Returns:
        Self for chaining
    """
    self.limit_val = limit
    return self

offset 

offset(offset)

Set query offset.

Parameters:

Name Type Description Default
offset int

Number of results to skip

 required

Returns:

Type Description
Query

Self for chaining
Source code in type_bridge/query/__init__.py 
126
127
128
129
130
131
132
133
134
135
136
def offset(self, offset: int) -> Query:
    """Set query offset.

    Args:
        offset: Number of results to skip

    Returns:
        Self for chaining
    """
    self.offset_val = offset
    return self

sort 

sort(variable, direction='asc')

Add sorting to the query.

Parameters:

Name Type Description Default
variable str

Variable to sort by

 required
direction str

Sort direction ("asc" or "desc")

 'asc'

Returns:

Type Description
Query

Self for chaining
Example

Query().match("$p isa person").fetch("$p").sort("$p", "asc")

Source code in type_bridge/query/__init__.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def sort(self, variable: str, direction: str = "asc") -> Query:
    """Add sorting to the query.

    Args:
        variable: Variable to sort by
        direction: Sort direction ("asc" or "desc")

    Returns:
        Self for chaining

    Example:
        Query().match("$p isa person").fetch("$p").sort("$p", "asc")
    """
    if direction not in ("asc", "desc"):
        raise ValueError(f"Invalid sort direction: {direction}")
    self.sort_clauses.append((variable, direction))
    return self

build 

build()

Build the final TypeQL query string.

Returns:

Type Description
str

Complete TypeQL query
Source code in type_bridge/query/__init__.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def build(self) -> str:
    """Build the final TypeQL query string.

    Returns:
        Complete TypeQL query
    """
    logger.debug("Building TypeQL query")
    parts = []

    # Match clause
    if self.match_clause.patterns:
        parts.append(self.compiler.compile(self.match_clause))

    # Delete clause
    if self.delete_clause.statements:
        parts.append(self.compiler.compile(self.delete_clause))

    # Insert clause
    if self.insert_clause.statements:
        parts.append(self.compiler.compile(self.insert_clause))

    # Sort, offset, and limit modifiers (must come BEFORE fetch in TypeQL 3.x)
    modifier_parts = []
    if self.sort_clauses:
        sort_items = [f"{var} {dir}" for var, dir in self.sort_clauses]
        modifier_parts.append(f"sort {', '.join(sort_items)};")

    # Order matters: form modifiers, put offset then limit
    if self.offset_val is not None:
        modifier_parts.append(f"offset {self.offset_val};")
    if self.limit_val is not None:
        modifier_parts.append(f"limit {self.limit_val};")

    if modifier_parts:
        parts.append("\n".join(modifier_parts))

    # Fetch clause
    if self.fetch_clause.items:
        parts.append(self.compiler.compile(self.fetch_clause))

    query = "\n".join(parts)
    logger.debug(f"Built query: {query}")
    return query

__str__ 

__str__()

String representation of query.

Source code in type_bridge/query/__init__.py 
200
201
202
def __str__(self) -> str:
    """String representation of query."""
    return self.build()

QueryBuilder

Helper class for building raw TypeQL from installed generated models.

match_entity staticmethod 

match_entity(model_class, var='$e', **filters)

Create a match query for an entity.

Parameters:

Name Type Description Default
model_class type[GeneratedEntityProjection]

An exact class from an installed generated projection

 required
var str

Variable name to use

 '$e'
filters Any

Attribute filters (field_name: value)

 {}

Returns:

Type Description
Query

Query object
Source code in type_bridge/query/__init__.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
@staticmethod
def match_entity(
    model_class: type[GeneratedEntityProjection],
    var: str = "$e",
    **filters: Any,
) -> Query:
    """Create a match query for an entity.

    Args:
        model_class: An exact class from an installed generated projection
        var: Variable name to use
        filters: Attribute filters (field_name: value)

    Returns:
        Query object
    """
    logger.debug(
        f"QueryBuilder.match_entity: {model_class.__name__}, var={var}, filters={filters}"
    )
    query = Query()
    from type_bridge._runtime_projection import projected_query_builder_match_entity_for

    pattern = projected_query_builder_match_entity_for(model_class, var, filters)
    query.match(pattern)
    return query

insert_entity staticmethod 

insert_entity(instance, var='$e')

Create an insert query for an entity instance.

Parameters:

Name Type Description Default
instance GeneratedEntityProjection

An exact value from an installed generated projection

 required
var str

Variable name to use

 '$e'

Returns:

Type Description
Query

Query object
Source code in type_bridge/query/__init__.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@staticmethod
def insert_entity(instance: GeneratedEntityProjection, var: str = "$e") -> Query:
    """Create an insert query for an entity instance.

    Args:
        instance: An exact value from an installed generated projection
        var: Variable name to use

    Returns:
        Query object
    """
    logger.debug(f"QueryBuilder.insert_entity: {instance.__class__.__name__}, var={var}")
    query = Query()
    from type_bridge._runtime_projection import projected_query_builder_insert_entity_for

    insert_pattern = projected_query_builder_insert_entity_for(instance, var)
    query.insert(insert_pattern)
    return query

match_relation staticmethod 

match_relation(model_class, var='$r', role_players=None)

Create a match query for a relation.

Parameters:

Name Type Description Default
model_class type[GeneratedRelationProjection]

An exact class from an installed generated projection

 required
var str

Variable name to use

 '$r'
role_players dict[str, str] | None

Dict mapping role names to player variables

 None

Returns:

Type Description
Query

Query object

Raises:

Type Description
ValueError

If a role name is not defined in the model
Source code in type_bridge/query/__init__.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
@staticmethod
def match_relation(
    model_class: type[GeneratedRelationProjection],
    var: str = "$r",
    role_players: dict[str, str] | None = None,
) -> Query:
    """Create a match query for a relation.

    Args:
        model_class: An exact class from an installed generated projection
        var: Variable name to use
        role_players: Dict mapping role names to player variables

    Returns:
        Query object

    Raises:
        ValueError: If a role name is not defined in the model
    """
    logger.debug(
        f"QueryBuilder.match_relation: {model_class.__name__}, var={var}, "
        f"role_players={role_players}"
    )
    query = Query()
    from type_bridge._runtime_projection import projected_query_builder_match_relation_for

    pattern = projected_query_builder_match_relation_for(model_class, var, role_players)
    query.match(pattern)
    return query

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 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
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 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
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 
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
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 
574
575
576
577
578
579
580
581
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 
593
594
595
596
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 
598
599
600
601
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 
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
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 
638
639
640
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 
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
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 
667
668
669
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 
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
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 
696
697
698
699
700
701
702
703
704
705
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 
707
708
709
710
711
712
713
714
715
716
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 
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
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 
735
736
737
738
739
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 
741
742
743
744
745
746
747
748
749
750
751
752
753
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 
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
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 
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
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 
811
812
813
814
815
816
817
818
819
820
821
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 
823
824
825
826
827
828
829
830
831
832
833
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 
835
836
837
838
839
840
841
842
843
844
845
846
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 
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
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 
885
886
887
888
889
890
891
892
893
894
895
896
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

TransactionContext 

TransactionContext(db, tx_type)

Context manager for sharing a TypeDB transaction across operations.

Source code in type_bridge/session.py 
1045
1046
1047
1048
1049
1050
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 
1121
1122
1123
1124
1125
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 
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
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 
1143
1144
1145
1146
1147
1148
1149
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 
1151
1152
1153
1154
1155
1156
1157
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()

TransactionType

Bases: Enum

Rust-safe fallback transaction type for the default backend.

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

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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
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)

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 
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    reason: str = "",
    *,
    event: CrudEvent | None = None,
    hook: Any = None,
):
    self.reason = reason
    self.event = event
    self.hook = hook
    super().__init__(reason)

SchemaConflictError 

SchemaConflictError(diff, message=None)

Bases: Exception

A retained conflict diagnostic for existing compatibility callers.

Source code in type_bridge/migration/exceptions.py 
15
16
17
def __init__(self, diff: Any, message: str | None = None) -> None:
    self.diff = diff
    super().__init__(message or "Schema conflict detected")

has_breaking_changes 

has_breaking_changes()

Report whether the supplied historical diff has breaking members.

Source code in type_bridge/migration/exceptions.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
def has_breaking_changes(self) -> bool:
    """Report whether the supplied historical diff has breaking members."""
    return any(
        bool(getattr(self.diff, name, None))
        for name in (
            "removed_entities",
            "removed_relations",
            "removed_attributes",
            "modified_attributes",
            "modified_entities",
            "modified_relations",
        )
    )

SchemaValidationError

Bases: Exception

A retained schema-validation diagnostic.

SchemaIntrospector 

SchemaIntrospector(db)

Introspects TypeDB database schema.

Queries the database to discover all types, attributes, ownerships, and relations defined in the schema.

Example

introspector = SchemaIntrospector(db) schema = introspector.introspect()

print(f"Found {len(schema.entities)} entities") print(f"Found {len(schema.relations)} relations") print(f"Found {len(schema.attributes)} attributes")

Initialize introspector.

Parameters:

Name Type Description Default
db Database

Database connection

 required
Source code in type_bridge/migration/introspection.py 
364
365
366
367
368
369
370
def __init__(self, db: Database):
    """Initialize introspector.

    Args:
        db: Database connection
    """
    self.db = db

introspect_for_models 

introspect_for_models(models)

Introspect database schema for specific model types.

This is the TypeDB 3.x compatible approach that checks each model type individually instead of enumerating all types.

Parameters:

Name Type Description Default
models list[type[_QueryEntity] | type[_QueryRelation]]

List of model classes to check

 required

Returns:

Type Description
IntrospectedSchema

IntrospectedSchema with info about existing types
Source code in type_bridge/migration/introspection.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
def introspect_for_models(
    self, models: list[type[Entity] | type[Relation]]
) -> IntrospectedSchema:
    """Introspect database schema for specific model types.

    This is the TypeDB 3.x compatible approach that checks each
    model type individually instead of enumerating all types.

    Args:
        models: List of model classes to check

    Returns:
        IntrospectedSchema with info about existing types
    """
    from type_bridge._rust_runtime import introspect_schema
    from type_bridge.models.entity import _QueryEntity as Entity
    from type_bridge.models.relation import _QueryRelation as Relation

    schema = IntrospectedSchema()

    if not self.db.database_exists():
        logger.debug("Database does not exist, returning empty schema")
        return schema

    logger.info(f"Introspecting database schema for {len(models)} model types")

    live_schema = IntrospectedSchema.from_rust_schema_info(introspect_schema(self.db))
    schema = _filter_schema_for_models(live_schema, models, Entity, Relation)

    logger.info(
        f"Introspected: {len(schema.entities)} entities, "
        f"{len(schema.relations)} relations, "
        f"{len(schema.attributes)} attributes"
    )

    return schema

introspect 

introspect()

Query TypeDB schema and return structured info.

Returns:

Type Description
IntrospectedSchema

IntrospectedSchema with all discovered types
Source code in type_bridge/migration/introspection.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
def introspect(self) -> IntrospectedSchema:
    """Query TypeDB schema and return structured info.

    Returns:
        IntrospectedSchema with all discovered types
    """
    schema = IntrospectedSchema()

    if not self.db.database_exists():
        logger.debug("Database does not exist, returning empty schema")
        return schema

    logger.info("Introspecting database schema")

    from type_bridge._rust_runtime import introspect_schema

    schema = IntrospectedSchema.from_rust_schema_info(introspect_schema(self.db))

    logger.info(
        f"Introspected: {len(schema.entities)} entities, "
        f"{len(schema.relations)} relations, "
        f"{len(schema.attributes)} attributes"
    )

    return schema

create_driver_options 

create_driver_options(is_tls_enabled=False, *, tls_root_ca=None)

Create TypeDB driver options for the retained driver lines.

The same band map that drives the version gate drives option construction. Supported 3.11 and 3.12 drivers both use the positional DriverOptions(tls_config) form.

Parameters:

Name Type Description Default
is_tls_enabled bool

Whether to enable TLS for the driver connection.

 False
tls_root_ca str | PathLike[str] | None

Optional PEM root-CA path for an enabled TLS connection.

 None

Returns:

Type Description
DriverOptions

Configured DriverOptions instance.

Raises:

Type Description
UnsupportedVersionError

When the installed driver version is outside the supported range (no known band).
ValueError

When a root path is supplied while TLS is disabled or the root path is empty.
Source code in type_bridge/typedb_driver.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def create_driver_options(
    is_tls_enabled: bool = False,
    *,
    tls_root_ca: str | os.PathLike[str] | None = None,
) -> DriverOptions:
    """Create TypeDB driver options for the retained driver lines.

    The same band map that drives the version gate drives option construction.
    Supported 3.11 and 3.12 drivers both use the positional
    ``DriverOptions(tls_config)`` form.

    Args:
        is_tls_enabled: Whether to enable TLS for the driver connection.
        tls_root_ca: Optional PEM root-CA path for an enabled TLS connection.

    Returns:
        Configured ``DriverOptions`` instance.

    Raises:
        UnsupportedVersionError: When the installed driver version is outside
            the supported range (no known band).
        ValueError: When a root path is supplied while TLS is disabled or the
            root path is empty.
    """
    if tls_root_ca is not None and not is_tls_enabled:
        raise ValueError("tls_root_ca contradicts explicit TLS disablement")
    root_ca_path = None if tls_root_ca is None else _root_ca_path(tls_root_ca)

    installed = driver_version()
    b = _ensure_driver_interpreter_supported(installed)

    if b in (8, 9):
        driver_tls_config = _load_tls_config()
        if root_ca_path is not None:
            tls_config = driver_tls_config.enabled_with_root_ca(root_ca_path)
        elif is_tls_enabled:
            tls_config = driver_tls_config.enabled_with_native_root_ca()
        else:
            tls_config = driver_tls_config.disabled()
        return DriverOptions(tls_config)

    import type_bridge.version as _version  # local import avoids circular dependency

    min_v = _version.min_supported_version()
    max_l = _version.max_supported_line()
    if sys.version_info >= (3, 14):
        remediation = (
            "Install `type-bridge[typedb-driver]` (driver 3.12.3 on "
            "CPython 3.14) and target TypeDB 3.12."
        )
    else:
        remediation = (
            "Install `type-bridge[typedb-driver]` and select a driver line "
            "accepted by the target server."
        )
    raise _version.UnsupportedVersionError(
        f"Installed typedb-driver {installed!r} has no known protocol band; "
        f"supported driver lines fall in {min_v}–{max_l}.x. "
        f"{remediation}"
    )

__getattr__ 

__getattr__(name)

Load retained compatibility identities without importing authoring eagerly.

Source code in type_bridge/__init__.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __getattr__(name: str) -> Any:
    """Load retained compatibility identities without importing authoring eagerly."""
    target = _LAZY_EXPORTS.get(name)
    if target is None:
        from type_bridge.migration._archive_imports import archive_attribute

        return archive_attribute(__name__, name)

    from importlib import import_module

    module_name, attribute_name = target
    value = getattr(import_module(module_name), attribute_name)
    globals()[name] = value
    return value