gate(), import and tokens

Last updated 3 November 2026

gate() holds a block’s outcome until synced(), import_sqlite() loads a plain SQLite file into a new database, and set_token() replaces the token the process sends.

gate()

@contextlib.contextmanager
def gate(*dbs: UrsulaConnection, timeout: float | None = None) -> Iterator[None]: ...

with sqlite_ursula.gate(db):
    reply = handle(request)  # commits through db
send(reply)  # runs once Tonbo holds every commit the block made

Runs the block, then lets its outcome go on only once a synced() call made after the block ended returned on every database given, so nothing outside the process acts on a result, or an error, that rests on a commit Tonbo does not hold yet. A block that only read waits too: it may have read another connection’s pending commit.

• An UrsulaError the block raises passes at once, as it was raised, also when a library wrapped it (as __cause__, SQLAlchemy’s .orig, or in an ExceptionGroup): a refused write committed nothing, and a reset is what the wait would report. So does a SQLite error of a framework’s own connection that is a refused write or a reset.
• Every other Exception waits as a result does, then passes. Ctrl-C, a cancellation and SystemExit pass at once.
• The first failure on any database, such as a reset, ends the wait and raises instead of the outcome.
• Effects inside the block, such as an outbound call, are not gated: call db.synced() before them. To make such a call once across crashes, see Run side effects once.
• It raises invalid_argument before the block runs when given no database, or one open() did not make.

agate()

@contextlib.asynccontextmanager
async def agate(
    *dbs: UrsulaConnection,
    timeout: float | None = None,
) -> AsyncIterator[None]: ...

async with sqlite_ursula.agate(db, timeout=10):
    reply = await handle(request)  # commits through db
await send(reply)

gate() for asyncio, as async with: the same parameters and the same rules, and it waits on the event loop instead of blocking it. Cancelling the task ends the wait.

import_sqlite()

def import_sqlite(
    url: str,
    file: str | os.PathLike[str],
    *,
    token: str | None = None,
    timeout: float = DEFAULT_IMPORT_S,
) -> Imported: ...

Imports the plain SQLite database file into the database at url, as its first snapshot, page for page; then open() it as any other. On Tonbo, create the database first (see Manage databases): it starts empty, and the import fills it. VACUUM INTO, run on the source database with any SQLite, writes a copy with 4096-byte pages and no -wal:

PRAGMA page_size = 4096;
VACUUM INTO '/tmp/notes-import.db';

import os

import sqlite_ursula

done = sqlite_ursula.import_sqlite(
    os.environ["NOTES_URL"],
    "/tmp/notes-import.db",
    token=os.environ["TONBO_API_KEY"],
)
print(done.pages, done.bytes)
db = sqlite_ursula.open(os.environ["NOTES_URL"], token=os.environ["TONBO_API_KEY"])

• It blocks the calling thread until the import is done, for at most timeout. The token needs to write the database.
• The database must be empty. A stream that is not a database’s raises stream_format, and anything else stream_state, with nothing published.
• While it runs, opens of the database raise stream_state. An import that fails after it claimed the database leaves one that refuses opens and imports until it is deleted and created again, so an UrsulaUnavailableError from an import says fix.
• A snapshot larger than one request goes up in parts. How large a database may grow is on Limits.
• It raises UrsulaSetupError, such as invalid_argument for the file, local_setup when it cannot read it, protocol_mismatch or forked, and UrsulaUnavailableError when Tonbo did not answer in time (see Troubleshooting).
• It is not idempotent: once it succeeded, a second import of the database raises stream_state. After a failure that published nothing, call it again; after one that left the claim, delete the database, create it again and import again.

To export a database to a plain SQLite file, see Export a copy.

Imported

@dataclass(frozen=True)
class Imported:
    offset: int
    digest: str
    pages: int
    bytes: int

What the import published, the database’s first snapshot. A frozen dataclass.

set_token()

def set_token(token: str | None) -> None: ...

import sqlite_ursula

# Every 5 minutes, with an access token exchanged from an Agent's key:
sqlite_ursula.set_token(exchange(agent_key))

Sets the bearer token every request of the process carries from now on, for every database it has open, and sends again at once the requests Tonbo refused with 401. The process holds one token: the last set_token()’s, or the last open()’s with a token, applies. Use it for a token that changes, such as an access token exchanged from an Agent’s key, which lasts 10 minutes: exchange the key again before it expires and pass the new token here, or commits wait until you do (see The token exchange).

It returns None. Setting the same token again changes nothing. It raises forked in a forked child, and invalid_argument for a token that is not a string. A token Tonbo refuses shows as unauthorized in status()["replication"]["condition"], while commits wait.