type_bridge.proxy

proxy

Proxy database adapter for type-bridge server.

Routes queries through a type-bridge proxy server instead of connecting directly to TypeDB. Drop-in replacement for Database.

Usage

proxy = ProxyDatabase("http://localhost:8080", database="mydb") proxy.connect()

manager = Person.manager(proxy) manager.insert(alice) results = manager.all()

proxy.close()

ProxyError

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

Bases: Exception

Error returned by the proxy server.

Source code in type_bridge/proxy.py

def __init__(self, message: str, code: str = "UNKNOWN", details: Any = None):
    super().__init__(message)
    self.code = code
    self.details = details

ProxyTransaction

ProxyTransaction(proxy, tx_type)

Transaction-like object that executes queries via the proxy server.

For MVP, each query is an independent HTTP request (stateless).

Source code in type_bridge/proxy.py

def __init__(self, proxy: ProxyDatabase, tx_type: str):
    self._proxy = proxy
    self._tx_type = tx_type
    self._is_open = True

is_open property

is_open

Check if this proxy transaction is still open.

execute

execute(query)

Execute a query via the proxy server.

Source code in type_bridge/proxy.py

def execute(self, query: str) -> list[dict[str, Any]]:
    """Execute a query via the proxy server."""
    if not self._is_open:
        raise RuntimeError("ProxyTransaction is closed")
    return self._proxy._send_raw_query(query, self._tx_type)

commit

commit()

No-op for stateless MVP — each query is auto-committed by the server.

Source code in type_bridge/proxy.py

def commit(self) -> None:
    """No-op for stateless MVP — each query is auto-committed by the server."""
    logger.debug("ProxyTransaction.commit() (stateless, no-op)")
    self._is_open = False

rollback

rollback()

No-op for stateless MVP.

Source code in type_bridge/proxy.py

def rollback(self) -> None:
    """No-op for stateless MVP."""
    logger.debug("ProxyTransaction.rollback() (stateless, no-op)")
    self._is_open = False

close

close()

Close this proxy transaction.

Source code in type_bridge/proxy.py

def close(self) -> None:
    """Close this proxy transaction."""
    self._is_open = False

ProxyTransactionContext

ProxyTransactionContext(proxy, tx_type)

Context manager mimicking TransactionContext but routing through the proxy.

Source code in type_bridge/proxy.py

def __init__(self, proxy: ProxyDatabase, tx_type: str):
    self._proxy = proxy
    self._tx_type = tx_type
    self._tx: ProxyTransaction | None = None

transaction property

transaction

Underlying proxy transaction.

database property

database

Proxy database backing this context.

execute

execute(query)

Execute a query within this context.

Source code in type_bridge/proxy.py

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

commit

commit()

Commit the active transaction.

Source code in type_bridge/proxy.py

def commit(self) -> None:
    """Commit the active transaction."""
    self.transaction.commit()

rollback

rollback()

Rollback the active transaction.

Source code in type_bridge/proxy.py

def rollback(self) -> None:
    """Rollback the active transaction."""
    self.transaction.rollback()

manager

manager(model_cls)

Get a TypeDBManager bound to this proxy transaction.

Source code in type_bridge/proxy.py

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

    if issubclass(model_cls, (Entity, Relation)):
        return TypeDBManager(self.transaction, model_cls)
    raise TypeError("manager() expects an Entity or Relation subclass")

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

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

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

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

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

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

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)