Storage API

locallore.storage.db

SCHEMA_VERSION = 3 module-attribute

connect(path)

Source code in src/locallore/storage/db.py

def connect(path: Path) -> sqlite3.Connection:
    path.parent.mkdir(parents=True, exist_ok=True)
    connection = sqlite3.connect(path)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA journal_mode = WAL")
    connection.execute("PRAGMA foreign_keys = ON")
    connection.execute("PRAGMA busy_timeout = 30000")
    return connection

migrate(connection)

Source code in src/locallore/storage/db.py

def migrate(connection: sqlite3.Connection) -> None:
    connection.execute(
        "CREATE TABLE IF NOT EXISTS schema_migrations "
        "(version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL)"
    )
    applied = connection.execute(
        "SELECT 1 FROM schema_migrations WHERE version = ?", (SCHEMA_VERSION,)
    ).fetchone()
    if applied is None:
        schema = Path(__file__).with_name("schema.sql").read_text()
        connection.executescript(schema)
        connection.execute(
            "INSERT INTO schema_migrations(version, applied_at) VALUES (?, ?)",
            (SCHEMA_VERSION, datetime.now(UTC).isoformat()),
        )
    connection.commit()