Add schema migrations for index

src/password_manager/entry_management.py

@@ -27,6 +27,7 @@ from typing import Optional, Tuple, Dict, Any, List
 from pathlib import Path
 
 from termcolor import colored
+from password_manager.migrations import LATEST_VERSION
 
 from password_manager.vault import Vault
 from utils.file_lock import exclusive_lock
@@ -61,12 +62,12 @@ class EntryManager:
                 return data
             except Exception as e:
                 logger.error(f"Failed to load index: {e}")
-                return {"passwords": {}}
+                return {"schema_version": LATEST_VERSION, "passwords": {}}
         else:
             logger.info(
                 f"Index file '{self.index_file}' not found. Initializing new password database."
             )
-            return {"passwords": {}}
+            return {"schema_version": LATEST_VERSION, "passwords": {}}
 
     def _save_index(self, data: Dict[str, Any]) -> None:
         try:

src/password_manager/migrations.py

@@ -0,0 +1,43 @@
+"""Schema migration helpers for password index files."""
+
+from __future__ import annotations
+
+from typing import Callable, Dict
+
+MIGRATIONS: Dict[int, Callable[[dict], dict]] = {}
+
+
+def migration(from_ver: int) -> Callable[[Callable[[dict], dict]], Callable[[dict], dict]]:
+    """Register a migration function from *from_ver* to *from_ver* + 1."""
+
+    def decorator(func: Callable[[dict], dict]) -> Callable[[dict], dict]:
+        MIGRATIONS[from_ver] = func
+        return func
+
+    return decorator
+
+
+@migration(0)
+def _v0_to_v1(data: dict) -> dict:
+    """Inject schema_version field for initial upgrade."""
+    data["schema_version"] = 1
+    return data
+
+
+LATEST_VERSION = 1
+
+
+def apply_migrations(data: dict) -> dict:
+    """Upgrade *data* in-place to the latest schema version."""
+    current = data.get("schema_version", 0)
+    if current > LATEST_VERSION:
+        raise ValueError(f"Unsupported schema version {current}")
+
+    while current < LATEST_VERSION:
+        migrate = MIGRATIONS.get(current)
+        if migrate is None:
+            raise ValueError(f"No migration available from version {current}")
+        data = migrate(data)
+        current = data.get("schema_version", current + 1)
+
+    return data

src/password_manager/vault.py

@@ -29,8 +29,17 @@ class Vault:
 
     # ----- Password index helpers -----
     def load_index(self) -> dict:
-        """Return decrypted password index data as a dict."""
-        return self.encryption_manager.load_json_data(self.index_file)
+        """Return decrypted password index data as a dict, applying migrations."""
+        data = self.encryption_manager.load_json_data(self.index_file)
+        from .migrations import apply_migrations, LATEST_VERSION
+
+        version = data.get("schema_version", 0)
+        if version > LATEST_VERSION:
+            raise ValueError(
+                f"File schema version {version} is newer than supported {LATEST_VERSION}"
+            )
+        data = apply_migrations(data)
+        return data
 
     def save_index(self, data: dict) -> None:
         """Encrypt and write password index."""