aimbat.db

Database engine for the AIMBAT project file.

The engine is created from Settings.db_url (see aimbat._config), which defaults to a SQLite database at aimbat.db in the current working directory. The path can be overridden via environment variable or .env file:

AIMBAT_PROJECT=/path/to/project.db  # derives db_url automatically
AIMBAT_DB_URL=sqlite+pysqlite:///absolute/path/to/project.db  # explicit override

For SQLite connections, PRAGMA foreign_keys=ON is set automatically on every new connection to enforce referential integrity. If the project's schema is out of date, a one-time aimbat.core.SchemaStaleWarning is raised suggesting aimbat db upgrade; nothing is changed automatically. The same message is also logged at WARNING level (see aimbat.logger), so it remains visible in aimbat.log even if the on-screen warning is missed.

Third-party code using this module's engine directly can promote the warning to a hard error, either with warnings.simplefilter("error", SchemaStaleWarning) or by setting AIMBAT_STRICT_SCHEMA_CHECK=true (PYTHONWARNINGS/-W are not a reliable way to do this for this warning category; use AIMBAT_STRICT_SCHEMA_CHECK instead). AIMBAT's own CLI and TUI always treat a stale schema as a hard failure regardless of AIMBAT_STRICT_SCHEMA_CHECK; that setting only affects third-party code that imports engine from this module directly.

engine = create_engine(
    url=settings.db_url,
    echo=False,
    connect_args={"check_same_thread": False, "timeout": 30}
    if make_url(settings.db_url).get_backend_name()
    == "sqlite"
    else {},
)

set_sqlite_pragma(
    dbapi_connection: Connection,
    connection_record: ConnectionPoolEntry,
) -> None

68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

@event.listens_for(engine, "connect")
def set_sqlite_pragma(
    dbapi_connection: sqlite3.Connection, connection_record: ConnectionPoolEntry
) -> None:
    """Enable specific PRAGMAs for each new SQLite connection.

    Args:
        dbapi_connection: The raw DBAPI connection that was just opened.
        connection_record: The connection pool's record for the connection.
    """
    logger.debug(
        f"Configuring SQLite connection with: {', '.join(_SQLITE_PRAGMAS)}."
    )
    cursor = dbapi_connection.cursor()
    for pragma in _SQLITE_PRAGMAS:
        cursor.execute(pragma)
    cursor.close()