Writing a database provider¶
Xyberos ships one database provider — SQLite — and what makes that a design rather than a limitation is
that nothing above it depends on it. A service depends on DatabaseProvider, a protocol with three
members, so a different engine is a different registration rather than a different import, as long as
those three members keep their promises.
class DatabaseProvider(Protocol):
def transaction(self) -> AbstractAsyncContextManager[AsyncConnection]: ...
def retrying(self, operation, *, attempts=None, base_delay=None) -> Awaitable[T]: ...
async def health(self) -> None: ...
That is the contract. What follows is a provider written out longhand against it, and then the suite that decides whether it is telling the truth.
The provider¶
"""A database provider written out longhand, so every piece of the contract is visible.
``xyberos.database.contracts.DatabaseProvider`` asks for three things: a transaction context
manager that commits or rolls back, a ``retrying`` helper with a bounded budget, and a ``health``
check. This module implements exactly those, in about a hundred lines, over the same SQLAlchemy
async engine the shipped provider uses.
It is deliberately the shipped provider's *shape*: that shape is the contract, and a walkthrough
that hid it would be teaching a different framework. What the shipped provider adds is settings,
PRAGMA tuning for WAL, a documented error taxonomy, and a factory for dependency injection — all
conveniences on top of this, none of which change what a service may assume.
Swap ``sqlite+aiosqlite`` for another SQLAlchemy async dialect and the same class is a provider for
that database. Services depend on ``DatabaseProvider``, so none of them change.
"""
from __future__ import annotations
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, TypeVar
from sqlalchemy import event, text
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine
from xyberos.database import DatabaseProvider, TransactionOperation
from xyberos.errors import (
DatabaseConnectionError,
DatabaseError,
TransactionError,
TransientDatabaseError,
)
T = TypeVar("T")
#: What SQLite says when another writer holds the lock. Every provider maps its own dialect's
#: equivalent onto `TransientDatabaseError`, and this is the part of the contract retrying() reads.
LOCK_MESSAGES = ("database is locked", "database table is locked", "database schema is locked")
@dataclass(frozen=True)
class ProviderSettings:
"""Where this provider's database lives, and how patient it is about contention."""
path: Path
busy_timeout_ms: int = 5000
retry_attempts: int = 5
retry_base_delay: float = 0.05
@property
def url(self) -> str:
"""Return the SQLAlchemy async URL for this file."""
return f"sqlite+aiosqlite:///{self.path}"
class SqlProvider:
"""Owns one database file, its engine, and its transaction boundaries."""
def __init__(self, settings: ProviderSettings) -> None:
self.settings = settings
self._engine: AsyncEngine | None = None
@property
def engine(self) -> AsyncEngine:
"""Return the started engine, or report that the provider was never started."""
if self._engine is None:
raise DatabaseConnectionError("the provider has not been started")
return self._engine
async def start(self) -> SqlProvider:
"""Create the engine, wait up to the busy timeout for locks, and check connectivity."""
settings = self.settings
settings.path.parent.mkdir(parents=True, exist_ok=True)
engine = create_async_engine(settings.url)
event.listen(engine.sync_engine, "connect", _busy_timeout_hook(settings.busy_timeout_ms))
self._engine = engine
try:
await self.health()
except DatabaseError:
# A provider that fails its own readiness check must not leave an engine behind.
self._engine = None
await engine.dispose()
raise
return self
async def aclose(self) -> None:
"""Release the engine and its pooled connections."""
engine, self._engine = self._engine, None
if engine is not None:
await engine.dispose()
async def health(self) -> None:
"""Run a trivial query, so readiness reflects connectivity rather than intent."""
try:
async with self.engine.connect() as connection:
await connection.execute(text("SELECT 1"))
except SQLAlchemyError as error:
raise DatabaseConnectionError(f"health check failed: {error}") from error
@asynccontextmanager
async def transaction(self) -> AsyncIterator[AsyncConnection]:
"""Open a short-lived transaction, committing on success and rolling back on error."""
async with self.engine.connect() as connection:
transaction = await connection.begin()
try:
yield connection
except BaseException:
await _rollback_quietly(transaction)
raise
try:
await transaction.commit()
except SQLAlchemyError as error:
await _rollback_quietly(transaction)
raise _transaction_error(error) from error
async def retrying(
self,
operation: TransactionOperation[T],
*,
attempts: int | None = None,
base_delay: float | None = None,
) -> T:
"""Run an idempotent operation, retrying only the documented transient failure.
The budget is bounded and the delay grows linearly: a provider that waited forever would
turn a lock into an outage, which is the failure the contract exists to prevent.
"""
max_attempts = self.settings.retry_attempts if attempts is None else attempts
delay = self.settings.retry_base_delay if base_delay is None else base_delay
for attempt in range(1, max_attempts + 1):
try:
async with self.transaction() as connection:
return await operation(connection)
except Exception as error:
if not _is_lock_contention(error):
raise
if attempt >= max_attempts:
raise TransientDatabaseError(
f"the database stayed locked after {max_attempts} attempt(s)"
) from error
await asyncio.sleep(delay * attempt)
raise AssertionError("unreachable: the retry loop always returns or raises")
async def open_provider(path: Path) -> SqlProvider:
"""Create, start, and return a provider the caller owns and closes."""
return await SqlProvider(ProviderSettings(path=path)).start()
def _busy_timeout_hook(milliseconds: int) -> Any:
"""Return the connect hook that makes SQLite wait for a writer instead of failing at once."""
def hook(dbapi_connection: Any, _record: Any) -> None:
cursor = dbapi_connection.cursor()
cursor.execute(f"PRAGMA busy_timeout = {int(milliseconds)}")
cursor.close()
return hook
async def _rollback_quietly(transaction: Any) -> None:
"""Roll back, ignoring a failure that would mask the error being raised."""
try:
await transaction.rollback()
except Exception: # noqa: BLE001 - the original error is the one worth reporting
pass
def _transaction_error(error: SQLAlchemyError) -> DatabaseError:
"""Map a commit failure onto the contract's taxonomy."""
if _is_lock_contention(error):
return TransientDatabaseError(f"the transaction could not commit: {error}")
return TransactionError(f"the transaction could not commit: {error}")
def _is_lock_contention(error: BaseException) -> bool:
"""Whether an error is the documented transient write contention."""
if isinstance(error, TransientDatabaseError):
return True
if isinstance(error, OperationalError):
message = str(getattr(error, "orig", error)).lower()
return any(fragment in message for fragment in LOCK_MESSAGES)
return False
class CommitsOnFailure:
"""A provider that commits the work a failed transaction should have rolled back.
This is the mistake the contract exists to catch, and nothing about it looks wrong at a call
site: the context manager yields a connection and the work is written. The difference appears
only when the body raises, which is what the conformance suite does.
"""
def __init__(self, inner: DatabaseProvider) -> None:
self._inner = inner
@asynccontextmanager
async def transaction(self) -> AsyncIterator[AsyncConnection]:
async with self._inner.transaction() as connection:
try:
yield connection
except Exception:
await connection.commit()
raise
def retrying(
self,
operation: TransactionOperation[T],
*,
attempts: int | None = None,
base_delay: float | None = None,
) -> object:
return self._inner.retrying(operation, attempts=attempts, base_delay=base_delay)
async def health(self) -> None:
await self._inner.health()
It is about a hundred lines over the same SQLAlchemy async engine the shipped provider uses, because the
goal is to show the shape the contract asks for rather than to be clever. Swap sqlite+aiosqlite for
another SQLAlchemy async dialect and the same class is a provider for that database; services depend on
the protocol, so none of them change. The shipped provider adds settings, WAL tuning, a documented error
taxonomy, and a factory for dependency injection — conveniences on top of this, none of which change what
a service may assume.
The conformance suite¶
"""Walkthrough — a database provider, checked by the conformance suite.
Run it with:
python -m learn_extend.provider01_contract
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from learn_extend.sql_provider import CommitsOnFailure, open_provider
from xyberos.database.testing import assert_provider_contract
#: The suite builds a table of its own, so a provider needs nothing but a working database.
TABLE = "provider_walkthrough"
async def main() -> None:
"""Write a provider, prove it with the suite, and then prove the suite can refuse one."""
with tempfile.TemporaryDirectory() as directory:
provider = await open_provider(Path(directory) / "walkthrough.db")
checked = await assert_provider_contract(provider, table=TABLE)
print(f"the conformance suite passed {len(checked)} checks:")
for name in checked:
print(" -", name)
await provider.aclose()
print("aclose() released the engine; the application's shutdown is what calls it")
inner = await open_provider(Path(directory) / "broken.db")
try:
broken = CommitsOnFailure(inner)
try:
await assert_provider_contract(broken, table="broken_walkthrough") # type: ignore[arg-type]
except AssertionError as error:
print("a provider that commits rolled-back work is refused:", error)
finally:
await inner.aclose()
if __name__ == "__main__":
asyncio.run(main())
the conformance suite passed 6 checks:
- health() succeeds for a working provider
- transaction() commits when the body returns
- transaction() rolls back and re-raises when the body raises
- a failed transaction leaves the provider usable
- retrying() runs the operation and returns its value
- retrying() lets 4 concurrent writers finish
aclose() released the engine; the application's shutdown is what calls it
a provider that commits rolled-back work is refused: a transaction that raised did not roll its work back
assert_provider_contract(provider, table="provider_walkthrough") builds its own table, runs those six
checks, drops the table, and returns the checks it ran — so it is safe against a scratch database and
safe to run twice. The name is worth reading twice, though, because the second half of the transcript is
the part that makes the first half mean anything: the suite is also given a provider that commits work a
failed transaction should have rolled back, and it refuses it.
CommitsOnFailure is a small wrapper: it delegates everything to a working provider and commits in the
except branch. Nothing about it looks wrong at a call site — the context manager yields a connection
and the work is written — which is exactly why the suite asserts the rollback rather than trusting it.
This repository checks both directions for the shipped provider too
(tests/database/test_provider_contract.py): the real one passes, and two deliberately broken ones fail.
The semantics behind the checks¶
| Promise | What it means for an implementation | What breaks without it |
|---|---|---|
transaction() commits when the body returns |
the commit happens after the body, not around it | silent data loss that only appears under concurrency |
transaction() rolls back and re-raises on error |
the original exception reaches the caller unchanged | a caught error that looks handled while the write is half-applied |
| a failed transaction leaves the provider usable | the next transaction works on the same instances | a poisoned pool that fails every later request |
retrying() returns the operation's result |
the value is not swallowed by the retry loop | a None where a row was expected, found much later |
retrying() keeps a bounded budget |
a lock becomes an error, not an outage | an unbounded wait inside a request |
health() raises on a broken connection |
readiness reflects connectivity, not intent | a container that reports healthy while it cannot serve |
Two details in the walkthrough are contract terms rather than incidentals. The commit is wrapped so that
a commit failure is rolled back and mapped before it is raised: a transaction that cannot commit has
not committed, and a caller should not have to guess. And retrying retries only the documented
transient failure — write contention, mapped to TransientDatabaseError — with a budget that grows
linearly. Any other error is raised on the first attempt, because retrying a bug is how a small fault
becomes a long one.
Errors are mapped onto the framework's taxonomy rather than leaked as driver exceptions:
| Situation | The provider raises |
|---|---|
| the database cannot answer a trivial query | DatabaseConnectionError |
| a commit failed for a reason that will not improve | TransactionError |
| a write was refused because another writer holds the lock | TransientDatabaseError |
| anything else the driver reports | a DatabaseError subclass, never a raw driver error |
That mapping is what makes retrying meaningful: the only thing it knows how to retry is the one
transient case, and an application's own error handling sees a framework error rather than a vendor one.
Registering it¶
A provider is an app-scoped capability like any other: the application's shutdown closes it if it has an
aclose or close method, and every service that declares Annotated[DatabaseProvider, Inject()] gets
the same instance.
"""Walkthrough — giving an application a database provider of your own.
Run it with:
python -m learn_extend.provider02_registration
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from typing import Annotated
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncConnection
from learn_extend.sql_provider import open_provider
from xyberos import App, Inject
from xyberos.database import DatabaseProvider
from xyberos.errors import DatabaseConnectionError
#: The table this walkthrough uses. The provider knows nothing about it: that is the service's
#: business.
SCHEMA = "CREATE TABLE IF NOT EXISTS items (name TEXT NOT NULL)"
class Inventory:
"""A service that depends on the contract, not on the engine behind it."""
def __init__(self, database: Annotated[DatabaseProvider, Inject()]) -> None:
self._database = database
async def prepare(self) -> None:
"""Create the table this service owns, in a transaction of its own."""
async with self._database.transaction() as connection:
await connection.execute(text(SCHEMA))
async def add(self, name: str) -> None:
"""Write one row, in a transaction this service owns."""
async with self._database.transaction() as connection:
await connection.execute(
text("INSERT INTO items (name) VALUES (:name)"), {"name": name}
)
async def count(self) -> int:
"""Count the rows, retrying if another writer holds the lock."""
async def read(connection: AsyncConnection) -> int:
result = await connection.execute(text("SELECT COUNT(*) FROM items"))
return int(result.scalar_one())
return await self._database.retrying(read)
async def ready(self) -> bool:
"""Answer the question a readiness endpoint asks, rather than assuming it."""
try:
await self._database.health()
except DatabaseConnectionError:
return False
return True
async def main() -> None:
"""Register the provider eagerly, use it through a service, and let shutdown close it."""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "inventory.db"
app = App()
app.provide(DatabaseProvider, lambda: open_provider(path), eager=True)
app.provide(Inventory)
async with app.lifespan():
print(app.diagnostics().render().strip())
inventory = await app.resolve(Inventory)
await inventory.prepare()
for name in ("widget", "sprocket", "cog"):
await inventory.add(name)
print("a service counted its own rows:", await inventory.count())
print("the service asked the provider whether it was ready:", await inventory.ready())
print("after shutdown, the same question answers:", await inventory.ready())
if __name__ == "__main__":
asyncio.run(main())
providers:
DatabaseProvider [app, eager] from application
Inventory [app] from application
plugins:
(none)
a service counted its own rows: 3
the service asked the provider whether it was ready: True
after shutdown, the same question answers: False
Four things in that transcript are worth naming. The provider is registered under the protocol, not
under the class it was written as, so services depend on DatabaseProvider and a different engine is a
different registration rather than a different import. It is marked [app, eager], because
eager=True opened the database at startup: a misconfigured path is a failure to start rather than a
failure on the first request. Inventory opens its own transactions and delegates retrying, and knows
nothing else about the engine. And after app.lifespan() exits, the same readiness question answers
False — the application closed an engine it owns, which is the behaviour that keeps a test suite from
leaking connections across cases.
A provider is usually registered as an instance that is already started — app.provide(DatabaseProvider,
instance=my_provider, eager=True) — because that is the shape an application's own settings code
produces. The factory above is the same thing one step earlier: the runtime awaits the factory, so both
arrive at the same place, and app.with_overrides({DatabaseProvider: ...}) replaces either in a test
with no other change.
Before you publish one¶
- Pass the suite.
assert_provider_contractin your own test file, against the engine you support. - Refuse breakage. Keep the two broken-provider cases somewhere in your suite, the way
tests/database/test_provider_contract.pydoes. A conformance suite that passes for everything is evidence that means nothing. - Map errors, do not leak them. A driver exception that reaches application code ties it to your engine.
- Document the dialect specifics. Concurrency limits, isolation behaviour, and what "transient" means for your engine are yours to state; the reference lists the parts that differ most.
- Publish it as a package, and point readers at your provider from your own documentation. The framework will not discover it, and it will not pretend to.
Where to read next¶
- Writing a database provider — the exhaustive reference: the checklist, dialect differences, and what stays SQLite-specific.
- The provider registry — scopes, eager factories, and overrides.
- Database — what the shipped provider offers an application.