Merge pull request #115 from NHSDigital/DTOSS-12809-consolidate-utility-scripts

Consolidate database utility scripts

Author: steventux

Makefile

@@ -68,8 +68,11 @@ _install-uv:
 # ---------------------------------------------------------------------------
 #  Database backup
 # ---------------------------------------------------------------------------
-backup-db: # Backup sqlite databases @Operations
-	PYTHONPATH=src uv run python -m backup_main
+backup-dbs: # Backup sqlite databases @Operations
+	PYTHONPATH=src uv run python -c'import scripts.python.database; scripts.python.database.backup_databases()'
+
+reset-worklist: # Reset worklist database @Operations
+	PYTHONPATH=src uv run python -c'import scripts.python.database; scripts.python.database.reset_worklist_database()'
 
 # ---------------------------------------------------------------------------
 # Testing

scripts/bash/package_release.sh

@@ -61,7 +61,7 @@ echo ""
 
 # ── Validate required files ───────────────────────────────────────────────────
 
-REQUIRED_FILES=("src" "pyproject.toml" "uv.lock" "README.md")
+REQUIRED_FILES=("src/" "scripts/python/" "pyproject.toml" "uv.lock" "README.md")
 
 for item in "${REQUIRED_FILES[@]}"; do
     if [[ ! -e "${REPO_ROOT}/${item}" ]]; then
@@ -84,7 +84,7 @@ cd "$REPO_ROOT"
 # This ensures only tracked files are included, preventing accidental inclusion of
 # .env files, local secrets, or untracked files.
 echo "Creating archive from git tracked files..."
-git archive --format=zip -o "$ARTIFACT_PATH" HEAD src/ pyproject.toml uv.lock README.md
+git archive --format=zip -o "$ARTIFACT_PATH" HEAD "${REQUIRED_FILES[@]}"
 
 echo "Archive created successfully."
 

scripts/bat/backup_database.bat

@@ -3,4 +3,4 @@
 :: Adjust the Python path below if the deployment creates the venv elsewhere.
 @echo off
 set rootdir=%~1
-"%rootdir%\.venv\Scripts\python.exe" -m mwl_reset
+"%rootdir%\.venv\Scripts\python.exe" -c'import scripts.python.database; scripts.python.database.reset_worklist_database()'

scripts/python/database.py

@@ -0,0 +1,101 @@
+import logging
+import os
+import sqlite3
+from datetime import datetime
+
+logging.basicConfig(
+    level=os.getenv("LOG_LEVEL", "INFO").upper(),
+    format=os.getenv("LOG_FORMAT", "%(asctime)s - %(name)s - %(levelname)s - %(message)s"),
+)
+logger = logging.getLogger(__name__)
+
+
+def backup_database(db_path: str, backup_dir: str) -> str:
+    """
+    Backup a SQLite database to a timestamped file in backup_dir.
+
+    Returns the path of the backup file.
+    """
+    os.makedirs(backup_dir, exist_ok=True)
+    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    db_filename = os.path.basename(db_path)
+    backup_path = os.path.join(backup_dir, f"{timestamp}.{db_filename}.backup")
+    with sqlite3.connect(db_path) as conn:
+        with sqlite3.connect(backup_path) as backup_conn:
+            conn.backup(backup_conn)
+    return backup_path
+
+
+def backup_databases():
+    """
+    Backup all configured databases.
+
+    Environment variables:
+    PACS_DB_PATH:  Path to the PACS SQLite database
+    MWL_DB_PATH:   Path to the MWL SQLite database
+    BACKUP_PATH:   Directory for backups (default: ./backups)
+    """
+    pacs_db_path = os.getenv("PACS_DB_PATH")
+    mwl_db_path = os.getenv("MWL_DB_PATH")
+    backup_path = os.getenv("BACKUP_PATH", "./backups")
+
+    if not pacs_db_path and not mwl_db_path:
+        logger.warning("No database paths configured (PACS_DB_PATH or MWL_DB_PATH), skipping backup")
+        return
+
+    success = True
+
+    if pacs_db_path:
+        try:
+            pacs_backup_path = backup_database(pacs_db_path, backup_path)
+        except Exception as e:
+            logging.error(f"PACS backup failed: {e}")
+            success = False
+    else:
+        logging.info("PACS_DB_PATH not set, skipping PACS backup")
+
+    if mwl_db_path:
+        try:
+            mwl_backup_path = backup_database(mwl_db_path, backup_path)
+        except Exception as e:
+            logging.error(f"MWL backup failed: {e}")
+            success = False
+    else:
+        logging.info("MWL_DB_PATH not set, skipping MWL backup")
+
+    if success:
+        logger.info("All database backups completed successfully. Backup files:")
+        if pacs_db_path:
+            logger.info(f"  PACS backup: {pacs_backup_path}")
+        if mwl_db_path:
+            logger.info(f"  MWL backup: {mwl_backup_path}")
+
+
+def reset_worklist_database() -> int:
+    """
+    Backs up and clears the MWL database..
+
+    Environment variables:
+    MWL_DB_PATH:  Path to the MWL SQLite database (default: /var/lib/pacs/worklist.db)
+    BACKUP_PATH:  Directory for database backups (default: /var/lib/pacs/backups)
+    """
+    mwl_db_path = os.getenv("MWL_DB_PATH", "/var/lib/pacs/worklist.db")
+    backup_path = os.getenv("BACKUP_PATH", "/var/lib/pacs/backups")
+    count = 0
+
+    try:
+        path = backup_database(mwl_db_path, backup_path)
+        logger.info(f"Backup complete: {path}")
+    except Exception as e:
+        logger.error(f"MWL backup failed: {e}", exc_info=True)
+
+    try:
+        with sqlite3.connect(mwl_db_path) as conn:
+            cursor = conn.execute("DELETE FROM worklist_items")
+            conn.commit()
+            count = cursor.rowcount
+            logger.info(f"MWL reset complete: {count} items deleted")
+    except Exception as e:
+        logger.error(f"MWL clear failed: {e}", exc_info=True)
+
+    return count

src/backup_main.py

@@ -0,0 +1,79 @@
+import pytest
+import sqlite3
+import sys
+from pathlib import Path
+
+sys.path.append(f"{Path(__file__).parent.parent.parent}/scripts/python")
+
+from database import backup_database, reset_worklist_database
+
+
+@pytest.fixture
+def worklist_db(tmp_dir, monkeypatch):
+    db_path = f"{tmp_dir}/test_worklist.db"
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("CREATE TABLE worklist_items (accession_number TEXT PRIMARY KEY, patient_id TEXT)")
+        conn.execute("INSERT INTO worklist_items VALUES ('ACC001', 'P001')")
+        conn.execute("INSERT INTO worklist_items VALUES ('ACC002', 'P002')")
+        conn.commit()
+
+    monkeypatch.setenv("BACKUP_PATH", f"{tmp_dir}/backups")
+    monkeypatch.setenv("MWL_DB_PATH", db_path)
+    return db_path
+
+
+# Tests for backup_database
+def test_backup_creates_file(tmp_dir):
+    db_path = f"{tmp_dir}/test.db"
+    sqlite3.connect(db_path).close()
+
+    backup_path = backup_database(db_path, f"{tmp_dir}/backups")
+
+    assert Path(backup_path).exists()
+
+
+def test_backup_returns_timestamped_path(tmp_dir):
+    db_path = f"{tmp_dir}/test.db"
+    sqlite3.connect(db_path).close()
+
+    backup_path = backup_database(db_path, f"{tmp_dir}/backups")
+
+    assert backup_path.endswith(".db.backup")
+
+
+def test_backup_creates_backup_dir_if_missing(tmp_dir):
+    db_path = f"{tmp_dir}/test.db"
+    sqlite3.connect(db_path).close()
+    backup_dir = f"{tmp_dir}/backups/nested"
+
+    backup_path = backup_database(db_path, backup_dir)
+
+    assert Path(backup_path).exists()
+
+
+def test_backup_database_creates_backup(worklist_db):
+    backup_path = backup_database(worklist_db, str(Path(worklist_db).parent / "backups"))
+    assert Path(backup_path).exists()
+
+
+# Tests for reset_worklist_database
+def test_reset_worklist_database_deletes_all_rows(worklist_db):
+    reset_worklist_database()
+
+    with sqlite3.connect(worklist_db) as conn:
+        count = conn.execute("SELECT COUNT(*) FROM worklist_items").fetchone()[0]
+    assert count == 0
+
+
+def test_reset_worklist_database_returns_row_count(worklist_db):
+    assert reset_worklist_database() == 2
+
+
+def test_reset_worklist_database_returns_zero_when_empty(tmp_dir, monkeypatch):
+    db_path = f"{tmp_dir}/worklist.db"
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("CREATE TABLE worklist_items (accession_number TEXT PRIMARY KEY)")
+        conn.commit()
+
+    monkeypatch.setenv("MWL_DB_PATH", db_path)
+    assert reset_worklist_database() == 0