Lesson 3 — store data in SQLite¶

What you will build¶

A service that owns its SQL, and the transaction boundary that decides what survives. The service contains no HTTP type at all, which is the convention from lesson 12 of the Fundamentals path — and the reason it can be tested without a request.

What you need¶

Xyberos with the database extra:

PowerShellPOSIX

python -m pip install "xyberos[database]"

python -m pip install 'xyberos[database]'

That extra brings SQLAlchemy Core, aiosqlite, and Alembic. The provider is SQLite-specific, but the contract it satisfies — transaction(), retrying(), health() — is the seam another database would implement (ADR 0016).

The code¶

examples/learn_traditional/traditional03_database.py

"""Lesson 3 — storing data in SQLite.

Run it with:

    python -m learn_traditional.traditional03_database
"""

from __future__ import annotations

import asyncio
from typing import Annotated, Any

from sqlalchemy import Column, Integer, MetaData, String, Table, func, insert, select

from xyberos import App, Inject
from xyberos.database import temporary_database
from xyberos.database.contracts import DatabaseProvider

metadata = MetaData()

items = Table(
    "items",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String, nullable=False),
)


class ItemStore:
    """A service that owns its SQL, and nothing about HTTP."""

    def __init__(self, database: Annotated[DatabaseProvider, Inject()]) -> None:
        self._database = database

    async def create_schema(self) -> None:
        """Create the table, which a real application would do with a migration."""
        async with self._database.transaction() as connection:
            await connection.run_sync(metadata.create_all)

    async def add(self, name: str) -> None:
        """Insert one item."""
        async with self._database.transaction() as connection:
            await connection.execute(insert(items).values(name=name))

    async def add_then_fail(self, name: str) -> None:
        """Insert one item and then raise, so the transaction rolls back."""
        async with self._database.transaction() as connection:
            await connection.execute(insert(items).values(name=name))
            raise RuntimeError("something went wrong after the insert")

    async def all(self) -> list[dict[str, Any]]:
        """Return every item."""
        async with self._database.transaction() as connection:
            rows = (await connection.execute(select(items))).mappings().all()
        return [dict(row) for row in rows]

    async def count(self) -> int:
        """Return how many items are stored."""
        async with self._database.transaction() as connection:
            return int(
                (await connection.execute(select(func.count()).select_from(items))).scalar_one()
            )


async def main() -> None:
    async with temporary_database() as database:
        app = App()
        app.provide(DatabaseProvider, instance=database)
        app.provide(ItemStore)

        async with app.lifespan():
            store = await app.resolve(ItemStore)
            await store.create_schema()
            await store.add("notebook")
            print("rows:", await store.all())

            try:
                await store.add_then_fail("lost")
            except RuntimeError:
                pass
            print("rows after the rollback:", await store.count())


if __name__ == "__main__":
    asyncio.run(main())

How to run it¶

PowerShellPOSIX

$env:PYTHONPATH = "src;examples"
python -m learn_traditional.traditional03_database

PYTHONPATH=src:examples python -m learn_traditional.traditional03_database

What it prints¶

rows: [{'id': 1, 'name': 'notebook'}]
rows after the rollback: 1

The second line is the whole lesson. The module inserts lost inside a transaction and then raises; the count is still 1 afterwards, because the transaction rolled back rather than leaving a half-written row.

The transaction is the unit of work¶

Task	How
Read or write	`async with database.transaction() as connection:`
Retry a locked write	`await database.retrying(operation, attempts=5, base_delay=0.05)`
Create the schema	a migration, not `metadata.create_all` — the lesson does the latter to stay short
Test against SQLite	`async with temporary_database() as database:`

Two facts explain almost everything about working with SQLite here:

A transaction commits on success and rolls back when the body raises. That is why the failed insert left nothing behind, and why the service never needs its own try/except to stay consistent.
SQLite allows one writer at a time. Keep write transactions short and never hold one across a network call — a held lock is a stalled deployment, not a slow query.

retrying is opt-in and bounded, because only you know whether an operation is safe to run twice. If it is not, give it an idempotency key rather than hoping.

Common mistakes¶

Holding a transaction across other work. The body of the with block is the transaction; a model call or an HTTP request inside it serializes every other writer behind it.
Swallowing the error inside the transaction. try: ... except: pass around a failing write commits the transaction, because nothing raised out of the body. Let it raise, or the rollback never happens.
Using metadata.create_all in an application. It is in this lesson only to keep the file short; a real application applies a migration at deploy time, which is lesson 7.
Assuming the provider is the database. SqliteSettings and its PRAGMAs are SQLite-specific; a service that depends on DatabaseProvider is portable, and one that depends on SqliteSettings is not.

Exercise¶

Add a rename method, and a route that calls it.

Show the answer

examples/learn_traditional/traditional03_exercise.py

"""Lesson 3, exercise — a rename method, and the SQL it runs.

The answer to the lesson's exercise. The only new idea is `update`, and it is still inside one
transaction: a write that raises leaves the row as it was.
"""

from __future__ import annotations

import asyncio
from typing import Annotated, Any

from sqlalchemy import Column, Integer, MetaData, String, Table, insert, select, update

from xyberos import App, Inject
from xyberos.database import temporary_database
from xyberos.database.contracts import DatabaseProvider

metadata = MetaData()

items = Table(
    "items",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String, nullable=False),
)


class ItemStore:
    """A service that owns its SQL, and nothing about HTTP."""

    def __init__(self, database: Annotated[DatabaseProvider, Inject()]) -> None:
        self._database = database

    async def create_schema(self) -> None:
        """Create the table, which a real application would do with a migration."""
        async with self._database.transaction() as connection:
            await connection.run_sync(metadata.create_all)

    async def add(self, name: str) -> None:
        """Insert one item."""
        async with self._database.transaction() as connection:
            await connection.execute(insert(items).values(name=name))

    async def rename(self, item_id: int, name: str) -> None:
        """Rename one item."""
        async with self._database.transaction() as connection:
            await connection.execute(update(items).where(items.c.id == item_id).values(name=name))

    async def all(self) -> list[dict[str, Any]]:
        """Return every item."""
        async with self._database.transaction() as connection:
            rows = (await connection.execute(select(items))).mappings().all()
        return [dict(row) for row in rows]


async def main() -> None:
    """Store one item, rename it, and read it back."""
    async with temporary_database() as database:
        app = App()
        app.provide(DatabaseProvider, instance=database)
        app.provide(ItemStore)

        async with app.lifespan():
            store = await app.resolve(ItemStore)
            await store.create_schema()
            await store.add("notebook")
            await store.rename(1, "diary")
            print("after the rename:", await store.all())


if __name__ == "__main__":
    asyncio.run(main())

after the rename: [{'id': 1, 'name': 'diary'}]

The route is the same shape as lesson 2's: read the id from the path, read the body with `read_json_object`, and call the service. Nothing about the store changes, which is the payoff of keeping HTTP out of it.

Next¶

Lesson 4 puts this store behind routes — creating and listing users — and moves the SQL into a service the handlers only call.