typerdrive database modules

DatabaseManager stores database configuration and creates DatabaseSession instances. Sessions own connections, queries, transactions, and lifecycle. Database attachment applies registered migrations before command code runs. db setup creates an empty database without application migrations. db migrate requires an existing database and applies pending migrations without creating an absent file. See the database feature guide for lifecycle and command details.

typerdrive.database

Attributes

__all__ module-attribute

__all__ = [
    "DatabaseError",
    "DatabaseInitError",
    "DatabaseLifecycleError",
    "DatabaseManagementError",
    "DatabaseManager",
    "DatabaseSession",
    "DatabaseMigrationError",
    "DatabaseMigrationIntegrityError",
    "DatabaseQueryError",
    "DatabaseRollbackError",
    "DatabaseTransactionError",
    "Migration",
    "MigrationRegistry",
    "add_database_subcommand",
    "attach_database",
    "get_database_manager",
    "get_database_session",
]

Classes

DatabaseError

Bases: TyperdriveError

Source code in src/typerdrive/database/exceptions.py

class DatabaseError(TyperdriveError):
    exit_code: ExitCode = ExitCode.GENERAL_ERROR

Attributes

exit_code class-attribute instance-attribute

exit_code: ExitCode = GENERAL_ERROR

DatabaseInitError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseInitError(DatabaseError):
    pass

DatabaseLifecycleError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseLifecycleError(DatabaseError):
    pass

DatabaseManagementError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseManagementError(DatabaseError):
    pass

DatabaseManager

Source code in src/typerdrive/database/manager.py

class DatabaseManager:
    def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
        self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
        self.journal_mode = journal_mode.upper()
        if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
            raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")

    def create_session(self) -> DatabaseSession:
        return DatabaseSession(self.database_path, self.journal_mode)

Attributes

database_path instance-attribute

database_path = (
    Path(database_path)
    if database_path is not None
    else database_path
)

journal_mode instance-attribute

journal_mode = upper()

Functions

__init__

__init__(
    database_path: Path | str | None = None,
    journal_mode: str = "WAL",
)

Source code in src/typerdrive/database/manager.py

def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
    self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
    self.journal_mode = journal_mode.upper()
    if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
        raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")

create_session

create_session() -> DatabaseSession

Source code in src/typerdrive/database/manager.py

def create_session(self) -> DatabaseSession:
    return DatabaseSession(self.database_path, self.journal_mode)

DatabaseMigrationError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseMigrationError(DatabaseError):
    pass

DatabaseMigrationIntegrityError

Bases: DatabaseMigrationError

Source code in src/typerdrive/database/exceptions.py

class DatabaseMigrationIntegrityError(DatabaseMigrationError):
    pass

DatabaseQueryError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseQueryError(DatabaseError):
    pass

DatabaseRollbackError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseRollbackError(DatabaseError):
    pass

DatabaseSession

Source code in src/typerdrive/database/session.py

class DatabaseSession:
    def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
        self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
        self.journal_mode = journal_mode.upper()
        if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
            raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")
        self.connection: aiosqlite.Connection | None = None
        self._mode: str | None = None
        self._operation_lock = asyncio.Lock()
        self._transaction_owner: asyncio.Task[Any] | None = None
        self._sync_operation_lock = threading.RLock()
        self._sync_transaction_active = False
        self._sync_adapter = BlockingAsyncAdapter(self._sync_lifecycle)

    @property
    def is_open(self) -> bool:
        return self.connection is not None

    async def __aenter__(self):
        if self._mode is not None:
            raise DatabaseLifecycleError("Database session lifecycle is already active")
        self._mode = "async"
        try:
            await self._open()
        except BaseException:
            self._mode = None
            raise
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        await self._close()

    def __enter__(self):
        if self._mode is not None:
            raise DatabaseLifecycleError("Database session lifecycle is already active")
        ensure_sync_context(
            lambda: DatabaseLifecycleError(
                "Synchronous database lifecycles cannot run inside an event loop; use async with"
            )
        )
        self._mode = "sync"
        try:
            self._sync_adapter.__enter__()
        except BaseException:
            self._mode = None
            raise
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            return self._sync_adapter.__exit__(exc_type, exc_value, traceback)
        finally:
            self._mode = None

    @asynccontextmanager
    async def _sync_lifecycle(self) -> AsyncIterator["DatabaseSession"]:
        await self._open()
        try:
            yield self
        finally:
            await self._close()

    async def _open(self) -> None:
        try:
            self.database_path.parent.mkdir(parents=True, exist_ok=True)
            self.connection = await aiosqlite.connect(self.database_path, timeout=30)
            self.connection.row_factory = sqlite3.Row
            await self.connection.execute("PRAGMA busy_timeout=30000")
            await self._configure_journal_mode()
            await self.connection.execute("PRAGMA foreign_keys=ON")
        except BaseException as error:
            await self._close_connection()
            if isinstance(error, DatabaseInitError):
                raise
            DatabaseInitError.raise_from(f"Failed to initialize database at {self.database_path}", error)

    async def _configure_journal_mode(self) -> None:
        connection = DatabaseInitError.enforce_defined(self.connection, "Database connection is not open")
        deadline = asyncio.get_running_loop().time() + 30
        while True:
            try:
                cursor = await connection.execute(f"PRAGMA journal_mode={self.journal_mode}")
                row = DatabaseInitError.enforce_defined(
                    await cursor.fetchone(),
                    "SQLite returned no journal mode",
                )
                await cursor.close()
                actual_mode = str(row[0]).upper()
                DatabaseInitError.require_condition(
                    actual_mode == self.journal_mode,
                    f"SQLite did not apply requested journal mode {self.journal_mode}; actual mode is {actual_mode}",
                )
                return
            except sqlite3.OperationalError as error:
                if "locked" not in str(error).lower() or asyncio.get_running_loop().time() >= deadline:
                    raise
                await asyncio.sleep(0.05)

    async def _close(self) -> None:
        try:
            if self.connection is not None:
                await self._close_connection()
        finally:
            self._mode = None

    async def _close_connection(self) -> None:
        if self.connection is not None:
            connection = self.connection
            self.connection = None
            with DatabaseInitError.handle_errors(f"Failed to close database at {self.database_path}"):
                await asyncio.shield(connection.close())

    def _check_async(self) -> aiosqlite.Connection:
        if self._mode == "sync":
            DatabaseLifecycleError.require_condition(
                current_sync_adapter() is self._sync_adapter,
                "Database session uses its sync API for this lifecycle",
            )
        elif self._mode != "async":
            raise DatabaseLifecycleError("Enter the database session with 'async with' before using its async API")
        return DatabaseLifecycleError.enforce_defined(self.connection, "Database session connection is not open")

    def _check_sync(self) -> None:
        ensure_sync_context(
            lambda: DatabaseLifecycleError("Sync database wrappers cannot run inside an event loop; use the async API")
        )
        DatabaseLifecycleError.require_condition(
            self._mode != "async",
            "Database session uses its async API for this lifecycle",
        )
        DatabaseLifecycleError.require_condition(
            self._mode == "sync" and self.connection is not None,
            "Enter the database session with 'with' before using its sync API",
        )

    @asynccontextmanager
    async def _operation(self) -> AsyncIterator[aiosqlite.Connection]:
        connection = self._check_async()
        task = asyncio.current_task()
        if self._transaction_owner is task or current_sync_adapter() is self._sync_adapter:
            yield connection
        else:
            async with self._operation_lock:
                yield connection

    async def execute(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
        try:
            async with self._operation() as connection:
                cursor = await connection.execute(sql, parameters)
                rowcount = cursor.rowcount
                await cursor.close()
                if self._transaction_owner is not asyncio.current_task() and not self._sync_transaction_active:
                    await connection.commit()
                return rowcount
        except DatabaseLifecycleError:
            raise
        except Exception as error:
            raise DatabaseQueryError(f"Failed to execute database query: {error}") from error

    async def fetch_one(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
        try:
            async with self._operation() as connection:
                cursor = await connection.execute(sql, parameters)
                row = await cursor.fetchone()
                await cursor.close()
                return row
        except DatabaseLifecycleError:
            raise
        except Exception as error:
            raise DatabaseQueryError(f"Failed to fetch database row: {error}") from error

    async def fetch_all(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
        try:
            async with self._operation() as connection:
                cursor = await connection.execute(sql, parameters)
                rows = list(await cursor.fetchall())
                await cursor.close()
                return rows
        except DatabaseLifecycleError:
            raise
        except Exception as error:
            raise DatabaseQueryError(f"Failed to fetch database rows: {error}") from error

    async def fetch_value(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
        row = await self.fetch_one(sql, parameters)
        return None if row is None else row[0]

    async def _rollback(self, connection: aiosqlite.Connection) -> None:
        with DatabaseRollbackError.handle_errors("Failed to roll back database transaction"):
            await asyncio.shield(connection.rollback())

    @asynccontextmanager
    async def transaction(self, immediate: bool = False) -> AsyncIterator["DatabaseSession"]:
        connection = self._check_async()
        DatabaseLifecycleError.require_condition(
            self._transaction_owner is not asyncio.current_task(),
            "Nested database transactions are not supported",
        )
        await self._operation_lock.acquire()
        self._transaction_owner = asyncio.current_task()
        try:
            await connection.execute("BEGIN IMMEDIATE" if immediate else "BEGIN")
            yield self
            await connection.commit()
        except BaseException as error:
            await self._rollback(connection)
            if isinstance(error, sqlite3.Error):
                DatabaseTransactionError.raise_from("Database transaction failed", error)
            raise
        finally:
            self._transaction_owner = None
            self._operation_lock.release()

    def execute_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.execute, sql, parameters)

    def fetch_one_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.fetch_one, sql, parameters)

    def fetch_all_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.fetch_all, sql, parameters)

    def fetch_value_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.fetch_value, sql, parameters)

    @contextmanager
    def transaction_sync(self):
        self._check_sync()
        self._sync_operation_lock.acquire()
        try:
            self._check_sync()
            DatabaseLifecycleError.require_condition(
                not self._sync_transaction_active,
                "Nested database transactions are not supported",
            )
            self._sync_transaction_active = True
            try:
                self._sync_adapter.run(self._begin_sync_transaction)
                try:
                    yield self
                except BaseException:
                    self._sync_adapter.run(self._finish_sync_transaction, commit=False)
                    raise
                else:
                    self._sync_adapter.run(self._finish_sync_transaction, commit=True)
            finally:
                self._sync_transaction_active = False
        finally:
            self._sync_operation_lock.release()

    async def _begin_sync_transaction(self) -> None:
        connection = self._check_async()
        async with DatabaseTransactionError.handle_errors_async("Failed to begin database transaction"):
            await connection.execute("BEGIN")

    async def _finish_sync_transaction(self, commit: bool) -> None:
        connection = self._check_async()
        if commit:
            try:
                await connection.commit()
            except Exception as commit_error:
                try:
                    await self._rollback(connection)
                except DatabaseRollbackError as rollback_error:
                    failures = ExceptionGroup(
                        "Database transaction commit and rollback failed",
                        [commit_error, rollback_error],
                    )
                    raise DatabaseRollbackError(
                        f"Failed to commit database transaction: {commit_error}; rollback also failed"
                    ) from failures
                raise DatabaseTransactionError(
                    f"Failed to commit database transaction: {commit_error}"
                ) from commit_error
        else:
            await self._rollback(connection)

Attributes

connection instance-attribute

connection: Connection | None = None

database_path instance-attribute

database_path = (
    Path(database_path)
    if database_path is not None
    else database_path
)

is_open property

is_open: bool

journal_mode instance-attribute

journal_mode = upper()

Functions

__aenter__ async

__aenter__()

Source code in src/typerdrive/database/session.py

async def __aenter__(self):
    if self._mode is not None:
        raise DatabaseLifecycleError("Database session lifecycle is already active")
    self._mode = "async"
    try:
        await self._open()
    except BaseException:
        self._mode = None
        raise
    return self

__aexit__ async

__aexit__(exc_type, exc_value, traceback)

Source code in src/typerdrive/database/session.py

async def __aexit__(self, exc_type, exc_value, traceback):
    await self._close()

__enter__

__enter__()

Source code in src/typerdrive/database/session.py

def __enter__(self):
    if self._mode is not None:
        raise DatabaseLifecycleError("Database session lifecycle is already active")
    ensure_sync_context(
        lambda: DatabaseLifecycleError(
            "Synchronous database lifecycles cannot run inside an event loop; use async with"
        )
    )
    self._mode = "sync"
    try:
        self._sync_adapter.__enter__()
    except BaseException:
        self._mode = None
        raise
    return self

__exit__

__exit__(exc_type, exc_value, traceback)

Source code in src/typerdrive/database/session.py

def __exit__(self, exc_type, exc_value, traceback):
    try:
        return self._sync_adapter.__exit__(exc_type, exc_value, traceback)
    finally:
        self._mode = None

__init__

__init__(
    database_path: Path | str | None = None,
    journal_mode: str = "WAL",
)

Source code in src/typerdrive/database/session.py

def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
    self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
    self.journal_mode = journal_mode.upper()
    if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
        raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")
    self.connection: aiosqlite.Connection | None = None
    self._mode: str | None = None
    self._operation_lock = asyncio.Lock()
    self._transaction_owner: asyncio.Task[Any] | None = None
    self._sync_operation_lock = threading.RLock()
    self._sync_transaction_active = False
    self._sync_adapter = BlockingAsyncAdapter(self._sync_lifecycle)

execute async

execute(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> int

Source code in src/typerdrive/database/session.py

async def execute(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
    try:
        async with self._operation() as connection:
            cursor = await connection.execute(sql, parameters)
            rowcount = cursor.rowcount
            await cursor.close()
            if self._transaction_owner is not asyncio.current_task() and not self._sync_transaction_active:
                await connection.commit()
            return rowcount
    except DatabaseLifecycleError:
        raise
    except Exception as error:
        raise DatabaseQueryError(f"Failed to execute database query: {error}") from error

execute_sync

execute_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> int

Source code in src/typerdrive/database/session.py

def execute_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.execute, sql, parameters)

fetch_all async

fetch_all(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> list[sqlite3.Row]

Source code in src/typerdrive/database/session.py

async def fetch_all(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
    try:
        async with self._operation() as connection:
            cursor = await connection.execute(sql, parameters)
            rows = list(await cursor.fetchall())
            await cursor.close()
            return rows
    except DatabaseLifecycleError:
        raise
    except Exception as error:
        raise DatabaseQueryError(f"Failed to fetch database rows: {error}") from error

fetch_all_sync

fetch_all_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> list[sqlite3.Row]

Source code in src/typerdrive/database/session.py

def fetch_all_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.fetch_all, sql, parameters)

fetch_one async

fetch_one(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> sqlite3.Row | None

Source code in src/typerdrive/database/session.py

async def fetch_one(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
    try:
        async with self._operation() as connection:
            cursor = await connection.execute(sql, parameters)
            row = await cursor.fetchone()
            await cursor.close()
            return row
    except DatabaseLifecycleError:
        raise
    except Exception as error:
        raise DatabaseQueryError(f"Failed to fetch database row: {error}") from error

fetch_one_sync

fetch_one_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> sqlite3.Row | None

Source code in src/typerdrive/database/session.py

def fetch_one_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.fetch_one, sql, parameters)

fetch_value async

fetch_value(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> Any

Source code in src/typerdrive/database/session.py

async def fetch_value(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
    row = await self.fetch_one(sql, parameters)
    return None if row is None else row[0]

fetch_value_sync

fetch_value_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> Any

Source code in src/typerdrive/database/session.py

def fetch_value_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.fetch_value, sql, parameters)

transaction async

transaction(
    immediate: bool = False,
) -> AsyncIterator[DatabaseSession]

Source code in src/typerdrive/database/session.py

@asynccontextmanager
async def transaction(self, immediate: bool = False) -> AsyncIterator["DatabaseSession"]:
    connection = self._check_async()
    DatabaseLifecycleError.require_condition(
        self._transaction_owner is not asyncio.current_task(),
        "Nested database transactions are not supported",
    )
    await self._operation_lock.acquire()
    self._transaction_owner = asyncio.current_task()
    try:
        await connection.execute("BEGIN IMMEDIATE" if immediate else "BEGIN")
        yield self
        await connection.commit()
    except BaseException as error:
        await self._rollback(connection)
        if isinstance(error, sqlite3.Error):
            DatabaseTransactionError.raise_from("Database transaction failed", error)
        raise
    finally:
        self._transaction_owner = None
        self._operation_lock.release()

transaction_sync

transaction_sync()

Source code in src/typerdrive/database/session.py

@contextmanager
def transaction_sync(self):
    self._check_sync()
    self._sync_operation_lock.acquire()
    try:
        self._check_sync()
        DatabaseLifecycleError.require_condition(
            not self._sync_transaction_active,
            "Nested database transactions are not supported",
        )
        self._sync_transaction_active = True
        try:
            self._sync_adapter.run(self._begin_sync_transaction)
            try:
                yield self
            except BaseException:
                self._sync_adapter.run(self._finish_sync_transaction, commit=False)
                raise
            else:
                self._sync_adapter.run(self._finish_sync_transaction, commit=True)
        finally:
            self._sync_transaction_active = False
    finally:
        self._sync_operation_lock.release()

DatabaseTransactionError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseTransactionError(DatabaseError):
    pass

Migration dataclass

Source code in src/typerdrive/database/migrations.py

@dataclass(frozen=True)
class Migration:
    index: int
    migration_id: str
    description: str
    sql_text: str
    checksum: str

Attributes

checksum instance-attribute

checksum: str

description instance-attribute

description: str

index instance-attribute

index: int

migration_id instance-attribute

migration_id: str

sql_text instance-attribute

sql_text: str

Functions

__init__

__init__(
    index: int,
    migration_id: str,
    description: str,
    sql_text: str,
    checksum: str,
) -> None

MigrationRegistry

Source code in src/typerdrive/database/migrations.py

class MigrationRegistry:
    def __init__(self):
        self._migrations: list[Migration] = []

    @property
    def migrations(self) -> tuple[Migration, ...]:
        return tuple(self._migrations)

    def register_migration(self, desc: str, sql_text: str) -> Migration:
        index = len(self._migrations) + 1
        migration_id = f"{index:08d}{inflection.underscore(desc)}"
        checksum = hashlib.sha256(sql_text.encode("utf-8")).hexdigest()
        migration = Migration(index, migration_id, desc, sql_text, checksum)
        self._migrations.append(migration)
        return migration

    def validate(self) -> tuple[tuple[Migration, tuple[str, ...]], ...]:
        validated: list[tuple[Migration, tuple[str, ...]]] = []
        for migration in self._migrations:
            if not migration.description.strip():
                raise DatabaseMigrationError(f"Migration {migration.index} has an empty description")
            if not migration.sql_text.strip():
                raise DatabaseMigrationError(f"Migration {migration.index} has empty SQL text")
            validated.append((migration, self._split_sql(migration.sql_text)))
        return tuple(validated)

    async def apply(self, session: DatabaseSession) -> None:
        migrations = self.validate()
        await session.execute(
            "CREATE TABLE IF NOT EXISTS _typerdrive_migrations ("
            "migration_index INTEGER PRIMARY KEY, "
            "migration_id TEXT NOT NULL UNIQUE, "
            "checksum TEXT NOT NULL)"
        )
        async with session.transaction(immediate=True):
            await self._verify_history(session, migrations)
        for migration, statements in migrations:
            try:
                async with session.transaction(immediate=True):
                    await self._apply_migrations_in_transaction(session, ((migration, statements),), migrations)
            except DatabaseMigrationIntegrityError:
                raise
            except DatabaseMigrationError:
                raise
            except Exception as error:
                raise DatabaseMigrationError(f"Failed to apply migration {migration.migration_id}: {error}") from error

    async def _apply_in_transaction(self, session: DatabaseSession) -> None:
        migrations = self.validate()
        await session.execute(
            "CREATE TABLE _typerdrive_migrations ("
            "migration_index INTEGER PRIMARY KEY, "
            "migration_id TEXT NOT NULL UNIQUE, "
            "checksum TEXT NOT NULL)"
        )
        await self._apply_migrations_in_transaction(session, migrations, migrations)

    async def _apply_migrations_in_transaction(
        self,
        session: DatabaseSession,
        pending: tuple[tuple[Migration, tuple[str, ...]], ...],
        all_migrations: tuple[tuple[Migration, tuple[str, ...]], ...],
    ) -> None:
        for migration, statements in pending:
            applied = await self._verify_history(session, all_migrations)
            if migration.index in applied:
                continue
            try:
                for statement in statements:
                    await session.execute(statement)
                await session.execute(
                    "INSERT INTO _typerdrive_migrations (migration_index, migration_id, checksum) VALUES (?, ?, ?)",
                    (migration.index, migration.migration_id, migration.checksum),
                )
            except Exception as error:
                raise DatabaseMigrationError(f"Failed to apply migration {migration.migration_id}: {error}") from error

    async def get_status(self, database_path: Path | str) -> tuple[list[Migration], list[Migration]]:
        migrations = self.validate()
        path = Path(database_path)
        if not path.exists():
            return [], [migration for migration, _ in migrations]
        try:
            connection = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True)
            try:
                table = connection.execute(
                    "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '_typerdrive_migrations'"
                ).fetchone()
                rows = (
                    []
                    if table is None
                    else connection.execute(
                        "SELECT migration_index, migration_id, checksum "
                        "FROM _typerdrive_migrations ORDER BY migration_index"
                    ).fetchall()
                )
            finally:
                connection.close()
        except sqlite3.Error as error:
            raise DatabaseMigrationError(f"Failed to inspect database migration status: {error}") from error
        applied = self._validate_rows(rows, migrations)
        applied_migrations = [migration for migration, _ in migrations if migration.index in applied]
        pending_migrations = [migration for migration, _ in migrations if migration.index not in applied]
        return applied_migrations, pending_migrations

    async def _verify_history(
        self, session: DatabaseSession, migrations: tuple[tuple[Migration, tuple[str, ...]], ...]
    ) -> set[int]:
        rows = await session.fetch_all(
            "SELECT migration_index, migration_id, checksum FROM _typerdrive_migrations ORDER BY migration_index"
        )
        return self._validate_rows(rows, migrations)

    @staticmethod
    def _validate_rows(rows, migrations: tuple[tuple[Migration, tuple[str, ...]], ...]) -> set[int]:
        expected = {migration.index: migration for migration, _ in migrations}
        applied: set[int] = set()
        for expected_index, (index, migration_id, checksum) in enumerate(rows, start=1):
            if index != expected_index:
                raise DatabaseMigrationIntegrityError(f"Applied migration index sequence has a gap before {index}")
            migration = expected.get(index)
            if migration is None:
                raise DatabaseMigrationIntegrityError(f"Applied migration index {index} is no longer registered")
            if migration_id != migration.migration_id:
                raise DatabaseMigrationIntegrityError(
                    f"Migration {index} changed identity from {migration_id} to {migration.migration_id}"
                )
            if checksum != migration.checksum:
                raise DatabaseMigrationIntegrityError(f"Migration {migration_id} SQL checksum changed")
            applied.add(index)
        return applied

    @staticmethod
    def _split_sql(sql_text: str) -> tuple[str, ...]:
        statements: list[str] = []
        current = ""
        for character in sql_text:
            current += character
            if character == ";" and sqlite3.complete_statement(current):
                if current.strip():
                    statements.append(current.strip())
                current = ""
        if current.strip() and (not statements or MigrationRegistry._contains_sql_content(current)):
            raise DatabaseMigrationError("Migration SQL contains an incomplete statement")
        return tuple(statements)

    @staticmethod
    def _contains_sql_content(sql_text: str) -> bool:
        index = 0
        while index < len(sql_text):
            if sql_text[index].isspace():
                index += 1
            elif sql_text.startswith("--", index):
                newline = index + 2
                while newline < len(sql_text) and sql_text[newline] not in "\r\n":
                    newline += 1
                if newline == len(sql_text):
                    return False
                index = newline + 1
            elif sql_text.startswith("/*", index):
                end = sql_text.find("*/", index + 2)
                if end == -1:
                    return True
                index = end + 2
            else:
                return True
        return False

Attributes

migrations property

migrations: tuple[Migration, ...]

Functions

__init__

__init__()

Source code in src/typerdrive/database/migrations.py

def __init__(self):
    self._migrations: list[Migration] = []

apply async

apply(session: DatabaseSession) -> None

Source code in src/typerdrive/database/migrations.py

async def apply(self, session: DatabaseSession) -> None:
    migrations = self.validate()
    await session.execute(
        "CREATE TABLE IF NOT EXISTS _typerdrive_migrations ("
        "migration_index INTEGER PRIMARY KEY, "
        "migration_id TEXT NOT NULL UNIQUE, "
        "checksum TEXT NOT NULL)"
    )
    async with session.transaction(immediate=True):
        await self._verify_history(session, migrations)
    for migration, statements in migrations:
        try:
            async with session.transaction(immediate=True):
                await self._apply_migrations_in_transaction(session, ((migration, statements),), migrations)
        except DatabaseMigrationIntegrityError:
            raise
        except DatabaseMigrationError:
            raise
        except Exception as error:
            raise DatabaseMigrationError(f"Failed to apply migration {migration.migration_id}: {error}") from error

get_status async

get_status(
    database_path: Path | str,
) -> tuple[list[Migration], list[Migration]]

Source code in src/typerdrive/database/migrations.py

async def get_status(self, database_path: Path | str) -> tuple[list[Migration], list[Migration]]:
    migrations = self.validate()
    path = Path(database_path)
    if not path.exists():
        return [], [migration for migration, _ in migrations]
    try:
        connection = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True)
        try:
            table = connection.execute(
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '_typerdrive_migrations'"
            ).fetchone()
            rows = (
                []
                if table is None
                else connection.execute(
                    "SELECT migration_index, migration_id, checksum "
                    "FROM _typerdrive_migrations ORDER BY migration_index"
                ).fetchall()
            )
        finally:
            connection.close()
    except sqlite3.Error as error:
        raise DatabaseMigrationError(f"Failed to inspect database migration status: {error}") from error
    applied = self._validate_rows(rows, migrations)
    applied_migrations = [migration for migration, _ in migrations if migration.index in applied]
    pending_migrations = [migration for migration, _ in migrations if migration.index not in applied]
    return applied_migrations, pending_migrations

register_migration

register_migration(desc: str, sql_text: str) -> Migration

Source code in src/typerdrive/database/migrations.py

def register_migration(self, desc: str, sql_text: str) -> Migration:
    index = len(self._migrations) + 1
    migration_id = f"{index:08d}{inflection.underscore(desc)}"
    checksum = hashlib.sha256(sql_text.encode("utf-8")).hexdigest()
    migration = Migration(index, migration_id, desc, sql_text, checksum)
    self._migrations.append(migration)
    return migration

validate

validate() -> tuple[tuple[Migration, tuple[str, ...]], ...]

Source code in src/typerdrive/database/migrations.py

def validate(self) -> tuple[tuple[Migration, tuple[str, ...]], ...]:
    validated: list[tuple[Migration, tuple[str, ...]]] = []
    for migration in self._migrations:
        if not migration.description.strip():
            raise DatabaseMigrationError(f"Migration {migration.index} has an empty description")
        if not migration.sql_text.strip():
            raise DatabaseMigrationError(f"Migration {migration.index} has empty SQL text")
        validated.append((migration, self._split_sql(migration.sql_text)))
    return tuple(validated)

Functions

__getattr__

__getattr__(name: str) -> Any

Source code in src/typerdrive/database/__init__.py

def __getattr__(name: str) -> Any:
    if name in {"attach_database", "get_database_manager", "get_database_session"}:
        from typerdrive.database.attach import attach_database, get_database_manager, get_database_session

        return {
            "attach_database": attach_database,
            "get_database_manager": get_database_manager,
            "get_database_session": get_database_session,
        }[name]
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

add_database_subcommand

add_database_subcommand(
    cli: Typer, migration_registry: MigrationRegistry
) -> None

Source code in src/typerdrive/database/commands.py

def add_database_subcommand(cli: typer.Typer, migration_registry: MigrationRegistry) -> None:
    database_cli = typer.Typer(
        help="Inspect and manage the application database",
        invoke_without_command=True,
        no_args_is_help=True,
    )
    database_cli.command()(path)
    database_cli.command()(inspect)
    database_cli.command(name="status")(lambda: status(migration_registry))
    database_cli.command()(setup)
    database_cli.command(name="migrate")(lambda: migrate(migration_registry))
    database_cli.command(name="reset")(lambda: reset(migration_registry))
    cli.add_typer(database_cli, name="db")

attach_database

attach_database(
    migration_registry: MigrationRegistry | None = None,
    *,
    journal_mode: str = "WAL",
) -> Callable[
    [ContextFunction[P, T]], ContextFunction[P, T]
]

Source code in src/typerdrive/database/attach.py

def attach_database(
    migration_registry: MigrationRegistry | None = None,
    *,
    journal_mode: str = "WAL",
) -> Callable[[ContextFunction[P, T]], ContextFunction[P, T]]:
    registry = migration_registry if migration_registry is not None else MigrationRegistry()

    def decorate(func: ContextFunction[P, T]) -> ContextFunction[P, T]:
        manager_param_key: str | None = None
        session_param_key: str | None = None
        rewriter = SignatureRewriter(func)
        for key, hint in rewriter.hints.items():
            if hint is DatabaseManager:
                rewriter.cloak(key, Annotated[DatabaseManager | None, CloakingDevice])
                manager_param_key = key
            elif hint is DatabaseSession:
                rewriter.cloak(key, Annotated[DatabaseSession | None, CloakingDevice])
                session_param_key = key

        if manager_param_key is not None and session_param_key is not None:
            raise DatabaseLifecycleError("A command can inject either DatabaseManager or DatabaseSession, not both")

        if inspect.iscoroutinefunction(func):

            async def invoke_async(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs):
                manager = DatabaseManager(journal_mode=journal_mode)
                _bind_manager(ctx, manager)
                try:
                    async with manager.create_session() as migration_session:
                        await registry.apply(migration_session)
                    if manager_param_key is not None:
                        cast(dict[str, Any], kwargs)[manager_param_key] = manager
                        return await func(ctx, *args, **kwargs)
                    session = manager.create_session()
                    if session_param_key is not None:
                        cast(dict[str, Any], kwargs)[session_param_key] = session
                    async with session:
                        _bind_session(ctx, session)
                        return await func(ctx, *args, **kwargs)
                finally:
                    _clear_database(ctx)

            def wrapper(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs):
                return run_async(
                    invoke_async,
                    ctx,
                    *args,
                    running_loop_error=lambda: DatabaseLifecycleError(
                        "The synchronous Typer command boundary cannot run inside an event loop; "
                        "invoke the command synchronously"
                    ),
                    **kwargs,
                )
        else:

            def wrapper(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs):
                ensure_sync_context(
                    lambda: DatabaseLifecycleError(
                        "The synchronous Typer command boundary cannot run inside an event loop; "
                        "invoke the command synchronously"
                    )
                )
                manager = DatabaseManager(journal_mode=journal_mode)
                _bind_manager(ctx, manager)
                try:
                    with manager.create_session() as migration_session:
                        run_on_current_sync_adapter(registry.apply, migration_session)
                    if manager_param_key is not None:
                        cast(dict[str, Any], kwargs)[manager_param_key] = manager
                        result = func(ctx, *args, **kwargs)
                        if inspect.isawaitable(result):

                            async def await_result():
                                return await cast(Awaitable[Any], result)

                            return run_async(await_result)
                        return result
                    session = manager.create_session()
                    if session_param_key is not None:
                        cast(dict[str, Any], kwargs)[session_param_key] = session
                    with session:
                        _bind_session(ctx, session)
                        result = func(ctx, *args, **kwargs)
                        if inspect.isawaitable(result):

                            async def await_result():
                                return await cast(Awaitable[Any], result)

                            return run_on_current_sync_adapter(await_result)
                        return result
                finally:
                    _clear_database(ctx)

        rewriter.apply(wrapper)
        return wrapper

    return decorate

get_database_manager

get_database_manager(ctx: Context) -> DatabaseManager

Source code in src/typerdrive/database/attach.py

def get_database_manager(ctx: typer.Context) -> DatabaseManager:
    with DatabaseError.handle_errors(
        "Database manager is not bound to the context. Use the @attach_database() decorator"
    ):
        manager: Any = from_context(ctx, "database_manager")
    return DatabaseError.ensure_type(
        manager,
        DatabaseManager,
        "Item in user context at `database_manager` was not a DatabaseManager",
    )

get_database_session

get_database_session(ctx: Context) -> DatabaseSession

Source code in src/typerdrive/database/attach.py

def get_database_session(ctx: typer.Context) -> DatabaseSession:
    with DatabaseError.handle_errors(
        "Database session is not bound to the context. Use the @attach_database() decorator"
    ):
        session: Any = from_context(ctx, "database_session")
    return DatabaseError.ensure_type(
        session,
        DatabaseSession,
        "Item in user context at `database_session` was not a DatabaseSession",
    )

typerdrive.database.attach

Attributes

ContextFunction module-attribute

ContextFunction = Callable[Concatenate[Context, P], T]

P module-attribute

P = ParamSpec('P')

T module-attribute

T = TypeVar('T')

Classes

Functions

attach_database

attach_database(
    migration_registry: MigrationRegistry | None = None,
    *,
    journal_mode: str = "WAL",
) -> Callable[
    [ContextFunction[P, T]], ContextFunction[P, T]
]

Source code in src/typerdrive/database/attach.py

def attach_database(
    migration_registry: MigrationRegistry | None = None,
    *,
    journal_mode: str = "WAL",
) -> Callable[[ContextFunction[P, T]], ContextFunction[P, T]]:
    registry = migration_registry if migration_registry is not None else MigrationRegistry()

    def decorate(func: ContextFunction[P, T]) -> ContextFunction[P, T]:
        manager_param_key: str | None = None
        session_param_key: str | None = None
        rewriter = SignatureRewriter(func)
        for key, hint in rewriter.hints.items():
            if hint is DatabaseManager:
                rewriter.cloak(key, Annotated[DatabaseManager | None, CloakingDevice])
                manager_param_key = key
            elif hint is DatabaseSession:
                rewriter.cloak(key, Annotated[DatabaseSession | None, CloakingDevice])
                session_param_key = key

        if manager_param_key is not None and session_param_key is not None:
            raise DatabaseLifecycleError("A command can inject either DatabaseManager or DatabaseSession, not both")

        if inspect.iscoroutinefunction(func):

            async def invoke_async(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs):
                manager = DatabaseManager(journal_mode=journal_mode)
                _bind_manager(ctx, manager)
                try:
                    async with manager.create_session() as migration_session:
                        await registry.apply(migration_session)
                    if manager_param_key is not None:
                        cast(dict[str, Any], kwargs)[manager_param_key] = manager
                        return await func(ctx, *args, **kwargs)
                    session = manager.create_session()
                    if session_param_key is not None:
                        cast(dict[str, Any], kwargs)[session_param_key] = session
                    async with session:
                        _bind_session(ctx, session)
                        return await func(ctx, *args, **kwargs)
                finally:
                    _clear_database(ctx)

            def wrapper(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs):
                return run_async(
                    invoke_async,
                    ctx,
                    *args,
                    running_loop_error=lambda: DatabaseLifecycleError(
                        "The synchronous Typer command boundary cannot run inside an event loop; "
                        "invoke the command synchronously"
                    ),
                    **kwargs,
                )
        else:

            def wrapper(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs):
                ensure_sync_context(
                    lambda: DatabaseLifecycleError(
                        "The synchronous Typer command boundary cannot run inside an event loop; "
                        "invoke the command synchronously"
                    )
                )
                manager = DatabaseManager(journal_mode=journal_mode)
                _bind_manager(ctx, manager)
                try:
                    with manager.create_session() as migration_session:
                        run_on_current_sync_adapter(registry.apply, migration_session)
                    if manager_param_key is not None:
                        cast(dict[str, Any], kwargs)[manager_param_key] = manager
                        result = func(ctx, *args, **kwargs)
                        if inspect.isawaitable(result):

                            async def await_result():
                                return await cast(Awaitable[Any], result)

                            return run_async(await_result)
                        return result
                    session = manager.create_session()
                    if session_param_key is not None:
                        cast(dict[str, Any], kwargs)[session_param_key] = session
                    with session:
                        _bind_session(ctx, session)
                        result = func(ctx, *args, **kwargs)
                        if inspect.isawaitable(result):

                            async def await_result():
                                return await cast(Awaitable[Any], result)

                            return run_on_current_sync_adapter(await_result)
                        return result
                finally:
                    _clear_database(ctx)

        rewriter.apply(wrapper)
        return wrapper

    return decorate

get_database_manager

get_database_manager(ctx: Context) -> DatabaseManager

Source code in src/typerdrive/database/attach.py

def get_database_manager(ctx: typer.Context) -> DatabaseManager:
    with DatabaseError.handle_errors(
        "Database manager is not bound to the context. Use the @attach_database() decorator"
    ):
        manager: Any = from_context(ctx, "database_manager")
    return DatabaseError.ensure_type(
        manager,
        DatabaseManager,
        "Item in user context at `database_manager` was not a DatabaseManager",
    )

get_database_session

get_database_session(ctx: Context) -> DatabaseSession

Source code in src/typerdrive/database/attach.py

def get_database_session(ctx: typer.Context) -> DatabaseSession:
    with DatabaseError.handle_errors(
        "Database session is not bound to the context. Use the @attach_database() decorator"
    ):
        session: Any = from_context(ctx, "database_session")
    return DatabaseError.ensure_type(
        session,
        DatabaseSession,
        "Item in user context at `database_session` was not a DatabaseSession",
    )

typerdrive.database.commands

Attributes

T module-attribute

T = TypeVar('T')

Classes

Functions

add_database_subcommand

add_database_subcommand(
    cli: Typer, migration_registry: MigrationRegistry
) -> None

Source code in src/typerdrive/database/commands.py

def add_database_subcommand(cli: typer.Typer, migration_registry: MigrationRegistry) -> None:
    database_cli = typer.Typer(
        help="Inspect and manage the application database",
        invoke_without_command=True,
        no_args_is_help=True,
    )
    database_cli.command()(path)
    database_cli.command()(inspect)
    database_cli.command(name="status")(lambda: status(migration_registry))
    database_cli.command()(setup)
    database_cli.command(name="migrate")(lambda: migrate(migration_registry))
    database_cli.command(name="reset")(lambda: reset(migration_registry))
    cli.add_typer(database_cli, name="db")

database_path

database_path() -> Path

Source code in src/typerdrive/database/commands.py

def database_path() -> Path:
    return get_typerdrive_config().database_path

inspect

inspect() -> None

Source code in src/typerdrive/database/commands.py

@handle_errors("Failed to inspect database", handle_exc_class=DatabaseError)
def inspect() -> None:
    db_path = database_path()
    if not db_path.exists():
        typer.echo(f"Database does not exist: {db_path}")
        return
    try:
        connection = sqlite3.connect(f"{db_path.resolve().as_uri()}?mode=ro", uri=True)
        try:
            sqlite_version = connection.execute("SELECT sqlite_version()").fetchone()[0]
            schema_version = connection.execute("PRAGMA schema_version").fetchone()[0]
            journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0]
        finally:
            connection.close()
    except sqlite3.Error as error:
        raise DatabaseManagementError(f"Failed to inspect database metadata: {error}") from error
    typer.echo(f"Database: {db_path}")
    typer.echo("Exists: yes")
    typer.echo(f"SQLite version: {sqlite_version}")
    typer.echo(f"Schema version: {schema_version}")
    typer.echo(f"Journal mode: {journal_mode}")

migrate

migrate(registry: MigrationRegistry) -> None

Source code in src/typerdrive/database/commands.py

@handle_errors("Failed to apply database migrations", handle_exc_class=DatabaseError)
def migrate(registry: MigrationRegistry) -> None:
    db_path = database_path()
    if not db_path.exists():
        raise DatabaseManagementError(
            f"Database does not exist at {db_path}; run 'db setup' or invoke an attached command first"
        )

    async def apply_migrations() -> None:
        async with DatabaseManager(db_path).create_session() as session:
            await registry.apply(session)

    try:
        _run_cli_async(apply_migrations)
    except DatabaseError:
        raise
    except Exception as error:
        raise DatabaseManagementError(f"Failed to apply database migrations: {error}") from error
    typer.echo(f"Database migrations applied: {db_path}")

path

path() -> None

Source code in src/typerdrive/database/commands.py

def path() -> None:
    typer.echo(database_path())

reset

reset(registry: MigrationRegistry) -> None

Source code in src/typerdrive/database/commands.py

@handle_errors("Failed to reset database", handle_exc_class=DatabaseError)
def reset(registry: MigrationRegistry) -> None:
    db_path = database_path()
    typer.confirm("Reset the database and delete all application data?", abort=True)

    async def rebuild() -> None:
        async with DatabaseManager(db_path).create_session() as session:
            foreign_keys = await session.fetch_value("PRAGMA foreign_keys")
            await session.execute("PRAGMA foreign_keys=OFF")
            try:
                async with session.transaction(immediate=True):
                    objects = await session.fetch_all(
                        "SELECT type, name FROM sqlite_master "
                        "WHERE type IN ('view', 'trigger', 'table') AND name NOT LIKE 'sqlite_%' "
                        "ORDER BY CASE type WHEN 'view' THEN 0 WHEN 'trigger' THEN 1 ELSE 2 END"
                    )
                    for object_type, name in objects:
                        escaped_name = name.replace('"', '""')
                        await session.execute(f'DROP {object_type.upper()} IF EXISTS "{escaped_name}"')
                    await registry._apply_in_transaction(session)
            finally:
                await session.execute(f"PRAGMA foreign_keys={'ON' if foreign_keys else 'OFF'}")

    try:
        _run_cli_async(rebuild)
    except DatabaseError:
        raise
    except Exception as error:
        raise DatabaseManagementError(f"Failed to rebuild database after reset: {error}") from error
    typer.echo(f"Database reset: {db_path}")

setup

setup() -> None

Source code in src/typerdrive/database/commands.py

@handle_errors("Failed to set up database", handle_exc_class=DatabaseError)
def setup() -> None:
    db_path = database_path()
    if db_path.exists():
        typer.echo(f"Database is already set up: {db_path}")
        return

    async def initialize() -> None:
        async with DatabaseManager(db_path).create_session():
            pass

    try:
        _run_cli_async(initialize)
    except DatabaseError:
        raise
    except Exception as error:
        raise DatabaseManagementError(f"Failed to set up database at {db_path}: {error}") from error
    typer.echo(f"Database set up: {db_path}")

status

status(registry: MigrationRegistry) -> None

Source code in src/typerdrive/database/commands.py

@handle_errors("Failed to read database migration status", handle_exc_class=DatabaseError)
def status(registry: MigrationRegistry) -> None:
    applied, pending = _run_cli_async(registry.get_status, database_path())
    for migration in applied:
        typer.echo(f"applied {migration.migration_id}")
    for migration in pending:
        typer.echo(f"pending {migration.migration_id}")
    if not applied and not pending:
        typer.echo("All registered migrations are applied")

typerdrive.database.exceptions

Classes

DatabaseError

Bases: TyperdriveError

Source code in src/typerdrive/database/exceptions.py

class DatabaseError(TyperdriveError):
    exit_code: ExitCode = ExitCode.GENERAL_ERROR

Attributes

exit_code class-attribute instance-attribute

exit_code: ExitCode = GENERAL_ERROR

DatabaseInitError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseInitError(DatabaseError):
    pass

DatabaseLifecycleError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseLifecycleError(DatabaseError):
    pass

DatabaseManagementError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseManagementError(DatabaseError):
    pass

DatabaseMigrationError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseMigrationError(DatabaseError):
    pass

DatabaseMigrationIntegrityError

Bases: DatabaseMigrationError

Source code in src/typerdrive/database/exceptions.py

class DatabaseMigrationIntegrityError(DatabaseMigrationError):
    pass

DatabaseQueryError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseQueryError(DatabaseError):
    pass

DatabaseRollbackError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseRollbackError(DatabaseError):
    pass

DatabaseTransactionError

Bases: DatabaseError

Source code in src/typerdrive/database/exceptions.py

class DatabaseTransactionError(DatabaseError):
    pass

typerdrive.database.manager

Classes

DatabaseManager

Source code in src/typerdrive/database/manager.py

class DatabaseManager:
    def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
        self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
        self.journal_mode = journal_mode.upper()
        if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
            raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")

    def create_session(self) -> DatabaseSession:
        return DatabaseSession(self.database_path, self.journal_mode)

Attributes

database_path instance-attribute

database_path = (
    Path(database_path)
    if database_path is not None
    else database_path
)

journal_mode instance-attribute

journal_mode = upper()

Functions

__init__

__init__(
    database_path: Path | str | None = None,
    journal_mode: str = "WAL",
)

Source code in src/typerdrive/database/manager.py

def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
    self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
    self.journal_mode = journal_mode.upper()
    if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
        raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")

create_session

create_session() -> DatabaseSession

Source code in src/typerdrive/database/manager.py

def create_session(self) -> DatabaseSession:
    return DatabaseSession(self.database_path, self.journal_mode)

Functions

typerdrive.database.session

Classes

DatabaseSession

Source code in src/typerdrive/database/session.py

class DatabaseSession:
    def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
        self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
        self.journal_mode = journal_mode.upper()
        if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
            raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")
        self.connection: aiosqlite.Connection | None = None
        self._mode: str | None = None
        self._operation_lock = asyncio.Lock()
        self._transaction_owner: asyncio.Task[Any] | None = None
        self._sync_operation_lock = threading.RLock()
        self._sync_transaction_active = False
        self._sync_adapter = BlockingAsyncAdapter(self._sync_lifecycle)

    @property
    def is_open(self) -> bool:
        return self.connection is not None

    async def __aenter__(self):
        if self._mode is not None:
            raise DatabaseLifecycleError("Database session lifecycle is already active")
        self._mode = "async"
        try:
            await self._open()
        except BaseException:
            self._mode = None
            raise
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        await self._close()

    def __enter__(self):
        if self._mode is not None:
            raise DatabaseLifecycleError("Database session lifecycle is already active")
        ensure_sync_context(
            lambda: DatabaseLifecycleError(
                "Synchronous database lifecycles cannot run inside an event loop; use async with"
            )
        )
        self._mode = "sync"
        try:
            self._sync_adapter.__enter__()
        except BaseException:
            self._mode = None
            raise
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            return self._sync_adapter.__exit__(exc_type, exc_value, traceback)
        finally:
            self._mode = None

    @asynccontextmanager
    async def _sync_lifecycle(self) -> AsyncIterator["DatabaseSession"]:
        await self._open()
        try:
            yield self
        finally:
            await self._close()

    async def _open(self) -> None:
        try:
            self.database_path.parent.mkdir(parents=True, exist_ok=True)
            self.connection = await aiosqlite.connect(self.database_path, timeout=30)
            self.connection.row_factory = sqlite3.Row
            await self.connection.execute("PRAGMA busy_timeout=30000")
            await self._configure_journal_mode()
            await self.connection.execute("PRAGMA foreign_keys=ON")
        except BaseException as error:
            await self._close_connection()
            if isinstance(error, DatabaseInitError):
                raise
            DatabaseInitError.raise_from(f"Failed to initialize database at {self.database_path}", error)

    async def _configure_journal_mode(self) -> None:
        connection = DatabaseInitError.enforce_defined(self.connection, "Database connection is not open")
        deadline = asyncio.get_running_loop().time() + 30
        while True:
            try:
                cursor = await connection.execute(f"PRAGMA journal_mode={self.journal_mode}")
                row = DatabaseInitError.enforce_defined(
                    await cursor.fetchone(),
                    "SQLite returned no journal mode",
                )
                await cursor.close()
                actual_mode = str(row[0]).upper()
                DatabaseInitError.require_condition(
                    actual_mode == self.journal_mode,
                    f"SQLite did not apply requested journal mode {self.journal_mode}; actual mode is {actual_mode}",
                )
                return
            except sqlite3.OperationalError as error:
                if "locked" not in str(error).lower() or asyncio.get_running_loop().time() >= deadline:
                    raise
                await asyncio.sleep(0.05)

    async def _close(self) -> None:
        try:
            if self.connection is not None:
                await self._close_connection()
        finally:
            self._mode = None

    async def _close_connection(self) -> None:
        if self.connection is not None:
            connection = self.connection
            self.connection = None
            with DatabaseInitError.handle_errors(f"Failed to close database at {self.database_path}"):
                await asyncio.shield(connection.close())

    def _check_async(self) -> aiosqlite.Connection:
        if self._mode == "sync":
            DatabaseLifecycleError.require_condition(
                current_sync_adapter() is self._sync_adapter,
                "Database session uses its sync API for this lifecycle",
            )
        elif self._mode != "async":
            raise DatabaseLifecycleError("Enter the database session with 'async with' before using its async API")
        return DatabaseLifecycleError.enforce_defined(self.connection, "Database session connection is not open")

    def _check_sync(self) -> None:
        ensure_sync_context(
            lambda: DatabaseLifecycleError("Sync database wrappers cannot run inside an event loop; use the async API")
        )
        DatabaseLifecycleError.require_condition(
            self._mode != "async",
            "Database session uses its async API for this lifecycle",
        )
        DatabaseLifecycleError.require_condition(
            self._mode == "sync" and self.connection is not None,
            "Enter the database session with 'with' before using its sync API",
        )

    @asynccontextmanager
    async def _operation(self) -> AsyncIterator[aiosqlite.Connection]:
        connection = self._check_async()
        task = asyncio.current_task()
        if self._transaction_owner is task or current_sync_adapter() is self._sync_adapter:
            yield connection
        else:
            async with self._operation_lock:
                yield connection

    async def execute(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
        try:
            async with self._operation() as connection:
                cursor = await connection.execute(sql, parameters)
                rowcount = cursor.rowcount
                await cursor.close()
                if self._transaction_owner is not asyncio.current_task() and not self._sync_transaction_active:
                    await connection.commit()
                return rowcount
        except DatabaseLifecycleError:
            raise
        except Exception as error:
            raise DatabaseQueryError(f"Failed to execute database query: {error}") from error

    async def fetch_one(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
        try:
            async with self._operation() as connection:
                cursor = await connection.execute(sql, parameters)
                row = await cursor.fetchone()
                await cursor.close()
                return row
        except DatabaseLifecycleError:
            raise
        except Exception as error:
            raise DatabaseQueryError(f"Failed to fetch database row: {error}") from error

    async def fetch_all(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
        try:
            async with self._operation() as connection:
                cursor = await connection.execute(sql, parameters)
                rows = list(await cursor.fetchall())
                await cursor.close()
                return rows
        except DatabaseLifecycleError:
            raise
        except Exception as error:
            raise DatabaseQueryError(f"Failed to fetch database rows: {error}") from error

    async def fetch_value(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
        row = await self.fetch_one(sql, parameters)
        return None if row is None else row[0]

    async def _rollback(self, connection: aiosqlite.Connection) -> None:
        with DatabaseRollbackError.handle_errors("Failed to roll back database transaction"):
            await asyncio.shield(connection.rollback())

    @asynccontextmanager
    async def transaction(self, immediate: bool = False) -> AsyncIterator["DatabaseSession"]:
        connection = self._check_async()
        DatabaseLifecycleError.require_condition(
            self._transaction_owner is not asyncio.current_task(),
            "Nested database transactions are not supported",
        )
        await self._operation_lock.acquire()
        self._transaction_owner = asyncio.current_task()
        try:
            await connection.execute("BEGIN IMMEDIATE" if immediate else "BEGIN")
            yield self
            await connection.commit()
        except BaseException as error:
            await self._rollback(connection)
            if isinstance(error, sqlite3.Error):
                DatabaseTransactionError.raise_from("Database transaction failed", error)
            raise
        finally:
            self._transaction_owner = None
            self._operation_lock.release()

    def execute_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.execute, sql, parameters)

    def fetch_one_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.fetch_one, sql, parameters)

    def fetch_all_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.fetch_all, sql, parameters)

    def fetch_value_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
        self._check_sync()
        with self._sync_operation_lock:
            self._check_sync()
            return self._sync_adapter.run(self.fetch_value, sql, parameters)

    @contextmanager
    def transaction_sync(self):
        self._check_sync()
        self._sync_operation_lock.acquire()
        try:
            self._check_sync()
            DatabaseLifecycleError.require_condition(
                not self._sync_transaction_active,
                "Nested database transactions are not supported",
            )
            self._sync_transaction_active = True
            try:
                self._sync_adapter.run(self._begin_sync_transaction)
                try:
                    yield self
                except BaseException:
                    self._sync_adapter.run(self._finish_sync_transaction, commit=False)
                    raise
                else:
                    self._sync_adapter.run(self._finish_sync_transaction, commit=True)
            finally:
                self._sync_transaction_active = False
        finally:
            self._sync_operation_lock.release()

    async def _begin_sync_transaction(self) -> None:
        connection = self._check_async()
        async with DatabaseTransactionError.handle_errors_async("Failed to begin database transaction"):
            await connection.execute("BEGIN")

    async def _finish_sync_transaction(self, commit: bool) -> None:
        connection = self._check_async()
        if commit:
            try:
                await connection.commit()
            except Exception as commit_error:
                try:
                    await self._rollback(connection)
                except DatabaseRollbackError as rollback_error:
                    failures = ExceptionGroup(
                        "Database transaction commit and rollback failed",
                        [commit_error, rollback_error],
                    )
                    raise DatabaseRollbackError(
                        f"Failed to commit database transaction: {commit_error}; rollback also failed"
                    ) from failures
                raise DatabaseTransactionError(
                    f"Failed to commit database transaction: {commit_error}"
                ) from commit_error
        else:
            await self._rollback(connection)

Attributes

connection instance-attribute

connection: Connection | None = None

database_path instance-attribute

database_path = (
    Path(database_path)
    if database_path is not None
    else database_path
)

is_open property

is_open: bool

journal_mode instance-attribute

journal_mode = upper()

Functions

__aenter__ async

__aenter__()

Source code in src/typerdrive/database/session.py

async def __aenter__(self):
    if self._mode is not None:
        raise DatabaseLifecycleError("Database session lifecycle is already active")
    self._mode = "async"
    try:
        await self._open()
    except BaseException:
        self._mode = None
        raise
    return self

__aexit__ async

__aexit__(exc_type, exc_value, traceback)

Source code in src/typerdrive/database/session.py

async def __aexit__(self, exc_type, exc_value, traceback):
    await self._close()

__enter__

__enter__()

Source code in src/typerdrive/database/session.py

def __enter__(self):
    if self._mode is not None:
        raise DatabaseLifecycleError("Database session lifecycle is already active")
    ensure_sync_context(
        lambda: DatabaseLifecycleError(
            "Synchronous database lifecycles cannot run inside an event loop; use async with"
        )
    )
    self._mode = "sync"
    try:
        self._sync_adapter.__enter__()
    except BaseException:
        self._mode = None
        raise
    return self

__exit__

__exit__(exc_type, exc_value, traceback)

Source code in src/typerdrive/database/session.py

def __exit__(self, exc_type, exc_value, traceback):
    try:
        return self._sync_adapter.__exit__(exc_type, exc_value, traceback)
    finally:
        self._mode = None

__init__

__init__(
    database_path: Path | str | None = None,
    journal_mode: str = "WAL",
)

Source code in src/typerdrive/database/session.py

def __init__(self, database_path: Path | str | None = None, journal_mode: str = "WAL"):
    self.database_path = Path(database_path) if database_path is not None else get_typerdrive_config().database_path
    self.journal_mode = journal_mode.upper()
    if self.journal_mode not in {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}:
        raise DatabaseInitError(f"Unsupported SQLite journal mode: {journal_mode}")
    self.connection: aiosqlite.Connection | None = None
    self._mode: str | None = None
    self._operation_lock = asyncio.Lock()
    self._transaction_owner: asyncio.Task[Any] | None = None
    self._sync_operation_lock = threading.RLock()
    self._sync_transaction_active = False
    self._sync_adapter = BlockingAsyncAdapter(self._sync_lifecycle)

execute async

execute(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> int

Source code in src/typerdrive/database/session.py

async def execute(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
    try:
        async with self._operation() as connection:
            cursor = await connection.execute(sql, parameters)
            rowcount = cursor.rowcount
            await cursor.close()
            if self._transaction_owner is not asyncio.current_task() and not self._sync_transaction_active:
                await connection.commit()
            return rowcount
    except DatabaseLifecycleError:
        raise
    except Exception as error:
        raise DatabaseQueryError(f"Failed to execute database query: {error}") from error

execute_sync

execute_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> int

Source code in src/typerdrive/database/session.py

def execute_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> int:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.execute, sql, parameters)

fetch_all async

fetch_all(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> list[sqlite3.Row]

Source code in src/typerdrive/database/session.py

async def fetch_all(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
    try:
        async with self._operation() as connection:
            cursor = await connection.execute(sql, parameters)
            rows = list(await cursor.fetchall())
            await cursor.close()
            return rows
    except DatabaseLifecycleError:
        raise
    except Exception as error:
        raise DatabaseQueryError(f"Failed to fetch database rows: {error}") from error

fetch_all_sync

fetch_all_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> list[sqlite3.Row]

Source code in src/typerdrive/database/session.py

def fetch_all_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> list[sqlite3.Row]:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.fetch_all, sql, parameters)

fetch_one async

fetch_one(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> sqlite3.Row | None

Source code in src/typerdrive/database/session.py

async def fetch_one(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
    try:
        async with self._operation() as connection:
            cursor = await connection.execute(sql, parameters)
            row = await cursor.fetchone()
            await cursor.close()
            return row
    except DatabaseLifecycleError:
        raise
    except Exception as error:
        raise DatabaseQueryError(f"Failed to fetch database row: {error}") from error

fetch_one_sync

fetch_one_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> sqlite3.Row | None

Source code in src/typerdrive/database/session.py

def fetch_one_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> sqlite3.Row | None:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.fetch_one, sql, parameters)

fetch_value async

fetch_value(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> Any

Source code in src/typerdrive/database/session.py

async def fetch_value(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
    row = await self.fetch_one(sql, parameters)
    return None if row is None else row[0]

fetch_value_sync

fetch_value_sync(
    sql: str, parameters: Sequence[Any] | Iterable[Any] = ()
) -> Any

Source code in src/typerdrive/database/session.py

def fetch_value_sync(self, sql: str, parameters: Sequence[Any] | Iterable[Any] = ()) -> Any:
    self._check_sync()
    with self._sync_operation_lock:
        self._check_sync()
        return self._sync_adapter.run(self.fetch_value, sql, parameters)

transaction async

transaction(
    immediate: bool = False,
) -> AsyncIterator[DatabaseSession]

Source code in src/typerdrive/database/session.py

@asynccontextmanager
async def transaction(self, immediate: bool = False) -> AsyncIterator["DatabaseSession"]:
    connection = self._check_async()
    DatabaseLifecycleError.require_condition(
        self._transaction_owner is not asyncio.current_task(),
        "Nested database transactions are not supported",
    )
    await self._operation_lock.acquire()
    self._transaction_owner = asyncio.current_task()
    try:
        await connection.execute("BEGIN IMMEDIATE" if immediate else "BEGIN")
        yield self
        await connection.commit()
    except BaseException as error:
        await self._rollback(connection)
        if isinstance(error, sqlite3.Error):
            DatabaseTransactionError.raise_from("Database transaction failed", error)
        raise
    finally:
        self._transaction_owner = None
        self._operation_lock.release()

transaction_sync

transaction_sync()

Source code in src/typerdrive/database/session.py

@contextmanager
def transaction_sync(self):
    self._check_sync()
    self._sync_operation_lock.acquire()
    try:
        self._check_sync()
        DatabaseLifecycleError.require_condition(
            not self._sync_transaction_active,
            "Nested database transactions are not supported",
        )
        self._sync_transaction_active = True
        try:
            self._sync_adapter.run(self._begin_sync_transaction)
            try:
                yield self
            except BaseException:
                self._sync_adapter.run(self._finish_sync_transaction, commit=False)
                raise
            else:
                self._sync_adapter.run(self._finish_sync_transaction, commit=True)
        finally:
            self._sync_transaction_active = False
    finally:
        self._sync_operation_lock.release()

Functions

typerdrive.database.migrations

Classes

Migration dataclass

Source code in src/typerdrive/database/migrations.py

@dataclass(frozen=True)
class Migration:
    index: int
    migration_id: str
    description: str
    sql_text: str
    checksum: str

Attributes

checksum instance-attribute

checksum: str

description instance-attribute

description: str

index instance-attribute

index: int

migration_id instance-attribute

migration_id: str

sql_text instance-attribute

sql_text: str

Functions

__init__

__init__(
    index: int,
    migration_id: str,
    description: str,
    sql_text: str,
    checksum: str,
) -> None

MigrationRegistry

Source code in src/typerdrive/database/migrations.py

class MigrationRegistry:
    def __init__(self):
        self._migrations: list[Migration] = []

    @property
    def migrations(self) -> tuple[Migration, ...]:
        return tuple(self._migrations)

    def register_migration(self, desc: str, sql_text: str) -> Migration:
        index = len(self._migrations) + 1
        migration_id = f"{index:08d}{inflection.underscore(desc)}"
        checksum = hashlib.sha256(sql_text.encode("utf-8")).hexdigest()
        migration = Migration(index, migration_id, desc, sql_text, checksum)
        self._migrations.append(migration)
        return migration

    def validate(self) -> tuple[tuple[Migration, tuple[str, ...]], ...]:
        validated: list[tuple[Migration, tuple[str, ...]]] = []
        for migration in self._migrations:
            if not migration.description.strip():
                raise DatabaseMigrationError(f"Migration {migration.index} has an empty description")
            if not migration.sql_text.strip():
                raise DatabaseMigrationError(f"Migration {migration.index} has empty SQL text")
            validated.append((migration, self._split_sql(migration.sql_text)))
        return tuple(validated)

    async def apply(self, session: DatabaseSession) -> None:
        migrations = self.validate()
        await session.execute(
            "CREATE TABLE IF NOT EXISTS _typerdrive_migrations ("
            "migration_index INTEGER PRIMARY KEY, "
            "migration_id TEXT NOT NULL UNIQUE, "
            "checksum TEXT NOT NULL)"
        )
        async with session.transaction(immediate=True):
            await self._verify_history(session, migrations)
        for migration, statements in migrations:
            try:
                async with session.transaction(immediate=True):
                    await self._apply_migrations_in_transaction(session, ((migration, statements),), migrations)
            except DatabaseMigrationIntegrityError:
                raise
            except DatabaseMigrationError:
                raise
            except Exception as error:
                raise DatabaseMigrationError(f"Failed to apply migration {migration.migration_id}: {error}") from error

    async def _apply_in_transaction(self, session: DatabaseSession) -> None:
        migrations = self.validate()
        await session.execute(
            "CREATE TABLE _typerdrive_migrations ("
            "migration_index INTEGER PRIMARY KEY, "
            "migration_id TEXT NOT NULL UNIQUE, "
            "checksum TEXT NOT NULL)"
        )
        await self._apply_migrations_in_transaction(session, migrations, migrations)

    async def _apply_migrations_in_transaction(
        self,
        session: DatabaseSession,
        pending: tuple[tuple[Migration, tuple[str, ...]], ...],
        all_migrations: tuple[tuple[Migration, tuple[str, ...]], ...],
    ) -> None:
        for migration, statements in pending:
            applied = await self._verify_history(session, all_migrations)
            if migration.index in applied:
                continue
            try:
                for statement in statements:
                    await session.execute(statement)
                await session.execute(
                    "INSERT INTO _typerdrive_migrations (migration_index, migration_id, checksum) VALUES (?, ?, ?)",
                    (migration.index, migration.migration_id, migration.checksum),
                )
            except Exception as error:
                raise DatabaseMigrationError(f"Failed to apply migration {migration.migration_id}: {error}") from error

    async def get_status(self, database_path: Path | str) -> tuple[list[Migration], list[Migration]]:
        migrations = self.validate()
        path = Path(database_path)
        if not path.exists():
            return [], [migration for migration, _ in migrations]
        try:
            connection = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True)
            try:
                table = connection.execute(
                    "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '_typerdrive_migrations'"
                ).fetchone()
                rows = (
                    []
                    if table is None
                    else connection.execute(
                        "SELECT migration_index, migration_id, checksum "
                        "FROM _typerdrive_migrations ORDER BY migration_index"
                    ).fetchall()
                )
            finally:
                connection.close()
        except sqlite3.Error as error:
            raise DatabaseMigrationError(f"Failed to inspect database migration status: {error}") from error
        applied = self._validate_rows(rows, migrations)
        applied_migrations = [migration for migration, _ in migrations if migration.index in applied]
        pending_migrations = [migration for migration, _ in migrations if migration.index not in applied]
        return applied_migrations, pending_migrations

    async def _verify_history(
        self, session: DatabaseSession, migrations: tuple[tuple[Migration, tuple[str, ...]], ...]
    ) -> set[int]:
        rows = await session.fetch_all(
            "SELECT migration_index, migration_id, checksum FROM _typerdrive_migrations ORDER BY migration_index"
        )
        return self._validate_rows(rows, migrations)

    @staticmethod
    def _validate_rows(rows, migrations: tuple[tuple[Migration, tuple[str, ...]], ...]) -> set[int]:
        expected = {migration.index: migration for migration, _ in migrations}
        applied: set[int] = set()
        for expected_index, (index, migration_id, checksum) in enumerate(rows, start=1):
            if index != expected_index:
                raise DatabaseMigrationIntegrityError(f"Applied migration index sequence has a gap before {index}")
            migration = expected.get(index)
            if migration is None:
                raise DatabaseMigrationIntegrityError(f"Applied migration index {index} is no longer registered")
            if migration_id != migration.migration_id:
                raise DatabaseMigrationIntegrityError(
                    f"Migration {index} changed identity from {migration_id} to {migration.migration_id}"
                )
            if checksum != migration.checksum:
                raise DatabaseMigrationIntegrityError(f"Migration {migration_id} SQL checksum changed")
            applied.add(index)
        return applied

    @staticmethod
    def _split_sql(sql_text: str) -> tuple[str, ...]:
        statements: list[str] = []
        current = ""
        for character in sql_text:
            current += character
            if character == ";" and sqlite3.complete_statement(current):
                if current.strip():
                    statements.append(current.strip())
                current = ""
        if current.strip() and (not statements or MigrationRegistry._contains_sql_content(current)):
            raise DatabaseMigrationError("Migration SQL contains an incomplete statement")
        return tuple(statements)

    @staticmethod
    def _contains_sql_content(sql_text: str) -> bool:
        index = 0
        while index < len(sql_text):
            if sql_text[index].isspace():
                index += 1
            elif sql_text.startswith("--", index):
                newline = index + 2
                while newline < len(sql_text) and sql_text[newline] not in "\r\n":
                    newline += 1
                if newline == len(sql_text):
                    return False
                index = newline + 1
            elif sql_text.startswith("/*", index):
                end = sql_text.find("*/", index + 2)
                if end == -1:
                    return True
                index = end + 2
            else:
                return True
        return False

Attributes

migrations property

migrations: tuple[Migration, ...]

Functions

__init__

__init__()

Source code in src/typerdrive/database/migrations.py

def __init__(self):
    self._migrations: list[Migration] = []

apply async

apply(session: DatabaseSession) -> None

Source code in src/typerdrive/database/migrations.py

async def apply(self, session: DatabaseSession) -> None:
    migrations = self.validate()
    await session.execute(
        "CREATE TABLE IF NOT EXISTS _typerdrive_migrations ("
        "migration_index INTEGER PRIMARY KEY, "
        "migration_id TEXT NOT NULL UNIQUE, "
        "checksum TEXT NOT NULL)"
    )
    async with session.transaction(immediate=True):
        await self._verify_history(session, migrations)
    for migration, statements in migrations:
        try:
            async with session.transaction(immediate=True):
                await self._apply_migrations_in_transaction(session, ((migration, statements),), migrations)
        except DatabaseMigrationIntegrityError:
            raise
        except DatabaseMigrationError:
            raise
        except Exception as error:
            raise DatabaseMigrationError(f"Failed to apply migration {migration.migration_id}: {error}") from error

get_status async

get_status(
    database_path: Path | str,
) -> tuple[list[Migration], list[Migration]]

Source code in src/typerdrive/database/migrations.py

async def get_status(self, database_path: Path | str) -> tuple[list[Migration], list[Migration]]:
    migrations = self.validate()
    path = Path(database_path)
    if not path.exists():
        return [], [migration for migration, _ in migrations]
    try:
        connection = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True)
        try:
            table = connection.execute(
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '_typerdrive_migrations'"
            ).fetchone()
            rows = (
                []
                if table is None
                else connection.execute(
                    "SELECT migration_index, migration_id, checksum "
                    "FROM _typerdrive_migrations ORDER BY migration_index"
                ).fetchall()
            )
        finally:
            connection.close()
    except sqlite3.Error as error:
        raise DatabaseMigrationError(f"Failed to inspect database migration status: {error}") from error
    applied = self._validate_rows(rows, migrations)
    applied_migrations = [migration for migration, _ in migrations if migration.index in applied]
    pending_migrations = [migration for migration, _ in migrations if migration.index not in applied]
    return applied_migrations, pending_migrations

register_migration

register_migration(desc: str, sql_text: str) -> Migration

Source code in src/typerdrive/database/migrations.py

def register_migration(self, desc: str, sql_text: str) -> Migration:
    index = len(self._migrations) + 1
    migration_id = f"{index:08d}{inflection.underscore(desc)}"
    checksum = hashlib.sha256(sql_text.encode("utf-8")).hexdigest()
    migration = Migration(index, migration_id, desc, sql_text, checksum)
    self._migrations.append(migration)
    return migration

validate

validate() -> tuple[tuple[Migration, tuple[str, ...]], ...]

Source code in src/typerdrive/database/migrations.py

def validate(self) -> tuple[tuple[Migration, tuple[str, ...]], ...]:
    validated: list[tuple[Migration, tuple[str, ...]]] = []
    for migration in self._migrations:
        if not migration.description.strip():
            raise DatabaseMigrationError(f"Migration {migration.index} has an empty description")
        if not migration.sql_text.strip():
            raise DatabaseMigrationError(f"Migration {migration.index} has empty SQL text")
        validated.append((migration, self._split_sql(migration.sql_text)))
    return tuple(validated)