Skip to content

SQLite databases for applications

The database feature gives a Typer application durable structured storage through SQLite without imposing an ORM or application schema. Applications own their SQL, queries, and data models.

Overview

typerdrive provides a DatabaseManager that owns one aiosqlite connection per lifecycle. The database defaults to <XDG_STATE_HOME>/<app_name>/db.sqlite, or ~/.local/state/<app_name>/db.sqlite when XDG_STATE_HOME is unset. The path is separate from cache and files storage.

The manager supports async SQL operations and transactions. Synchronous commands can use explicit wrappers backed by the same aiosqlite connection. A lifecycle uses either the async API or the sync wrappers, never both. Sync wrappers reject calls from a running event loop; use the async methods instead.

WAL is the default journal mode. DatabaseManager and attach_database accept another supported SQLite journal mode when the storage environment requires it.

Async SQL and transactions

Enter the manager explicitly for async work that outlives one command:

import asyncio

from typerdrive import DatabaseManager


async def main():
    async with DatabaseManager() as database:
        await database.execute("CREATE TABLE IF NOT EXISTS notes (body TEXT NOT NULL)")
        await database.execute("INSERT INTO notes (body) VALUES (?)", ("Hello",))
        rows = await database.fetch_all("SELECT body FROM notes")
        print(rows[0]["body"])


asyncio.run(main())

execute, fetch_one, fetch_all, and fetch_value accept parameterized SQL. Query methods return SQLite rows, which support integer and column-name indexing. SQLite and connection failures become database feature errors.

A transaction serializes operations from other tasks on the same manager. Other tasks wait until the transaction commits or rolls back, and their operations do not join the transaction. Normal exit commits. An exception or cancellation rolls back:

async with database.transaction():
    await database.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, source_id))
    await database.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, target_id))

SQLite remains responsible for coordinating independent connections and processes. Unresolvable locking and contention failures are reported as feature errors.

Synchronous commands

Use the sync wrappers inside a normal with lifecycle. They bridge to the manager's aiosqlite connection and keep it open for the whole command:

with DatabaseManager() as database:
    database.execute_sync("INSERT INTO notes (body) VALUES (?)", ("Hello",))
    note = database.fetch_value_sync("SELECT body FROM notes LIMIT 1")

The sync methods are execute_sync, fetch_one_sync, fetch_all_sync, fetch_value_sync, and transaction_sync. Calling them inside an active event loop raises DatabaseLifecycleError and points to the async API. Do not use both sets of methods during one lifecycle.

Attach the database to commands

@attach_database creates and closes a manager around each command, binds it to TyperdriveContext, and injects a DatabaseManager parameter. For an async command, the synchronous Typer callback boundary runs one top-level event loop and awaits the command before closing the manager. Synchronous command functions use the sync wrappers.

import typer

from typerdrive import DatabaseManager, MigrationRegistry, attach_database

cli = typer.Typer()
registry = MigrationRegistry()
registry.register_migration("create_notes", "CREATE TABLE notes (body TEXT NOT NULL)")


@cli.command()
@attach_database(registry)
async def add_note(ctx: typer.Context, database: DatabaseManager, body: str):
    await database.execute("INSERT INTO notes (body) VALUES (?)", (body,))

Use get_database_manager(ctx) when a command prefers context lookup over typed injection. Attachment automatically creates and configures an absent database, then applies every registered migration before the callback starts. For an existing database, it applies pending migrations. A failed or invalid migration prevents callback execution. The manager closes after success, error, or cancellation. The synchronous Typer boundary cannot run inside an existing event loop; invoke CLI commands synchronously and use explicit async lifecycles for async applications.

Register SQL migrations

The application chooses how to load SQL text and passes the text and a non-empty description to MigrationRegistry. Registration order determines migration order, starting at index 1. Each ID is generated by f"{idx:08d}{inflection.underscore(desc)}"; the registry also stores a SHA-256 checksum of the exact SQL text.

from importlib.resources import files

from typerdrive import MigrationRegistry

registry = MigrationRegistry()
registry.register_migration(
    "create_notes",
    files("my_app.migrations").joinpath("001_create_notes.sql").read_text(),
)

Register the full history in order and append new migrations. Applied indices must remain registered with the same identifier and SQL checksum. Removing a migration, changing its description, or editing its SQL fails integrity validation before later migrations run. Migrations are forward-only. Each migration and its tracking record are applied atomically; a failed migration rolls back its changes and leaves earlier successful migrations available for retry.

The registry accepts strings only. It does not load files, call Python migration functions, or define application schemas. The tracking table is owned by typerdrive; all other tables and SQL remain application-owned.

Database management commands

Register the optional management group with the same registry used by attach_database:

from typerdrive import add_database_subcommand

add_database_subcommand(cli, registry)

The commands are:

  • db path prints the configured path without creating it.
  • db inspect reports database existence, SQLite version, schema version, and journal mode without application rows.
  • db status lists applied and pending migrations without modifying the database.
  • db setup creates and configures an empty database only; it does not apply application migrations and is a no-op when the database already exists.
  • db migrate requires an existing database and applies pending registered migrations. It does not create an absent database.
  • db reset requires confirmation and transactionally rebuilds the schema and migration history in the existing database file using the currently registered migrations.

If the database file is absent, inspect reports that fact and status lists every registered migration as pending; neither command creates the file. Concurrent setup initializes through SQLite and does not apply migrations. A reset that is declined preserves data and history. Existing readers retain a consistent SQLite snapshot while reset runs; concurrent writers serialize through SQLite's busy timeout. Reset does not unlink the database or its WAL sidecars.

Reset replaces all application data. Back up any data that must survive before confirming it.

ORM layering

The database feature exposes direct SQL and SQLite rows. It does not provide an ORM, model layer, or application schema. An application may build its own repository layer or integrate an ORM while using the attached manager's connection.