feat: Database migration system for automatic user DB upgrades (Issue #181)

[boscorat]

Summary

Implements automatic detection and upgrade of user project.db databases when a new BSP version is released, as described in Issue #181.

What changed

New module: modules/db_migration.py

Database-agnostic migration logic reusable by openstan:

• fingerprint_data_scripts() -- SHA-512 hashes of all data/*.py files for change detection
• needs_upgrade(db_path) -- checks db_meta version/hashes against current BSP
• migrate_db(db_path) -- creates fresh DB, copies raw data + user objects, archives old DB, replaces with upgraded DB
• _get_user_objects(conn) -- detects user-added tables/views/triggers/indexes not in BSP's known schema
• _copy_user_objects(old_conn, new_conn, user_objects) -- copies user tables (schema + data), views, triggers, and indexes

Changes to existing modules

• create_project_db.py -- writes db_meta table with BSP version + script hashes on new DB creation
• database.py -- calls migrate_db() before _require_db() in update_db()
• mock_project_data.py -- writes db_meta with INSERT OR REPLACE for test DBs
• init.py -- exports needs_upgrade and migrate_db

Upgrade flow

1. update_db() calls migrate_db(db_path) which calls needs_upgrade(db_path)
2. If version/hashes differ: create temp DB with _upgrade suffix, copy raw tables, detect and copy user objects
3. Archive old DB to database_archive/project_v{old_version}.db
4. Promote temp DB to canonical name
5. On failure: warnings.warn() emitted, original DB preserved, BSP continues with existing DB

User object preservation

• User tables: schema + data copied via INSERT OR REPLACE
• User views: CREATE VIEW SQL replayed
• User triggers: CREATE TRIGGER SQL replayed
• User indexes: CREATE INDEX SQL replayed
• BSP objects identified by _BSP_OWNED_TABLES / _BSP_OWNED_VIEWS frozensets plus index table/name heuristic

Test coverage

22 new tests in tests/test_db_migration.py:

• TestFingerprintScripts (4 tests): deterministic, SHA-512, expected keys, no pycache
• TestNeedsUpgrade (5 tests): no meta, matching, different version, different hashes, nonexistent
• TestGetUserObjects (5 tests): BSP exclusion, user table/view/index detection
• TestMigrateDb (8 tests): no-op, user objects, raw data, archival, db_meta, failure, nonexistent, old DB without meta

All 261 tests pass.

Checklist

• ruff check -- clean (1 pre-existing page_crop unused import, not from this PR)
• ruff format -- all files unchanged
• pytest -v -- 261 passed
• Docs regenerated via scripts/generate_docs.py
• New exports added to all and docs/reference/python-api.md

[Copilot]

🟡 Changes recommended

There are confirmed correctness and API/docs issues (e.g., needs_upgrade() JSON decode crash path, internal SQLite object handling, and an unintended public API/doc removal) that should be fixed before approval.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR introduces an automatic SQLite project.db migration system that detects schema drift (via BSP version + data-script fingerprints) and upgrades user databases in-place while attempting to preserve user-created objects.

Changes:

• Added db_migration module with script fingerprinting, upgrade detection, user-object discovery, and a full migration/archival flow.
• Integrated migrations into the persistence path (update_db()), and wrote db_meta on database creation + mock-data seeding.
• Updated the public API exports and Python API reference docs to expose migration helpers.

File summaries

File	Description
tests/test_db_migration.py	Adds unit tests for fingerprinting, upgrade detection, user-object detection, and migration behavior.
src/bank_statement_parser/modules/db_migration.py	Implements the migration system (fingerprinting, needs_upgrade, migrate_db, user object preservation).
src/bank_statement_parser/modules/database.py	Calls migrate_db() before requiring/using the DB in update_db().
src/bank_statement_parser/data/mock_project_data.py	Writes db_meta (version + hashes) into test DBs seeded with mock data.
src/bank_statement_parser/data/create_project_db.py	Writes db_meta (version + hashes) when creating a new project DB.
src/bank_statement_parser/__init__.py	Exports migrate_db / needs_upgrade via the package public API.
docs/reference/python-api.md	Documents bsp.migrate_db() and bsp.needs_upgrade().

Review details

• Files reviewed: 7/7 changed files
• Comments generated: 10
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🟡 Changes recommended

The migration logic has a few correctness and robustness issues (e.g., invalid script_hashes handling, SQLite internal object detection, index ownership heuristic) that can cause upgrade failures or silent loss of user objects.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (6)

Previously missed (3) — in code that hasn't changed since the last review.

src/bank_statement_parser/modules/db_migration.py:221

• Index ownership detection is too aggressive: any index on a BSP-owned table with an idx_ prefix is treated as BSP-owned and skipped. A user-created index like idx_user_label on statement_heads would be silently dropped, contradicting the “user objects preserved” behavior.

            # An index is BSP-owned if it sits on a BSP-owned table and
            # follows the BSP naming convention (starts with ``idx_``).
            if tbl_name in _BSP_OWNED_TABLES and name.startswith("idx_"):
                continue
            user_objects["indexes"].append({"name": name, "sql": sql})

src/bank_statement_parser/data/create_project_db.py:204

• Writing db_meta with plain INSERT can fail with UNIQUE constraint failed: db_meta.key if main() is ever re-run against an existing DB (e.g. partial upgrade DB left behind). Using INSERT OR REPLACE makes this idempotent (and matches mock_project_data).

    conn.execute("INSERT INTO db_meta (key, value) VALUES ('bsp_version', ?)", (__version__,))
    conn.execute("INSERT INTO db_meta (key, value) VALUES ('script_hashes', ?)", (json.dumps(hashes),))

src/bank_statement_parser/modules/db_migration.py:82

• _DDL_DB_META is imported and used by other modules (create_project_db.py, mock_project_data.py), but its leading underscore marks it as private. Consider making it public (e.g. DDL_DB_META) or exposing a small write_db_meta(...) helper to avoid cross-module reliance on a private symbol.

_DDL_DB_META = """
CREATE TABLE IF NOT EXISTS db_meta (

src/bank_statement_parser/modules/db_migration.py:188

• needs_upgrade() will raise if db_meta.script_hashes contains invalid JSON (or a non-string), which can crash callers like update_db(). Treat unreadable hashes as “needs upgrade” instead of propagating the exception.

        stored_hashes = json.loads(meta.get("script_hashes", "{}"))
        current_hashes = fingerprint_data_scripts()
        return stored_hashes != current_hashes

src/bank_statement_parser/modules/db_migration.py:212

• _get_user_objects() currently treats SQLite internal objects (e.g. sqlite_sequence, sqlite_stat1) as user tables/indexes. Copying these can break migrations or create irrelevant objects in the upgraded DB; they should be excluded.

    for obj_type, name, tbl_name, sql in rows:
        if obj_type == "table" and name not in _BSP_OWNED_TABLES:
            user_objects["tables"].append({"name": name, "sql": sql})
        elif obj_type == "view" and name not in _BSP_OWNED_VIEWS:

src/bank_statement_parser/modules/db_migration.py:244

• In _copy_user_objects(), the column-name extraction via SELECT ... LIMIT 0).fetchall() is incorrect (it always returns an empty list) and is immediately overwritten by the PRAGMA result. Removing it avoids confusion and makes the intent clearer.

            cols = [desc[0] for desc in old_conn.execute(f'SELECT * FROM "{table_info["name"]}" LIMIT 0').fetchall() or []]
            # Re-fetch column names via PRAGMA for robustness
            pragma_rows = old_conn.execute(f'PRAGMA table_info("{table_info["name"]}")').fetchall()
            cols = [r[1] for r in pragma_rows]

• Files reviewed: 7/7 changed files
• Comments generated: 1
• Review effort level: Lite