feat: Implement application preferences with JSON persistence
Build and Push Docker Image / build-and-push (push) Has been cancelled

- Added preferences management in `preferences.py` with functions to load, save, get, set, and reset preferences.
- Introduced a configuration directory structure based on the operating system.
- Integrated preferences into the settings window, allowing users to reset settings and manage window geometry.
- Enhanced `search_filter.py` to support flexible date column names and improved filtering logic.
- Updated `settings_window.py` to include options for managing backup and configuration folder paths.
- Introduced an `UndoManager` class to handle undo actions for add/update/delete operations.
- Improved UIManager to support sorting in tree views and added a toast notification feature.
This commit is contained in:
William Valentin
2025-08-07 16:26:17 -07:00
parent 73498af138
commit 9372d6ef29
15 changed files with 1997 additions and 468 deletions
+221 -179
View File
@@ -1,63 +1,121 @@
"""Auto-save functionality for TheChart application."""
"""Auto-save and backup utilities for TheChart.
Provides two APIs:
New application API (used by main app):
AutoSaveManager(save_callback=callable, interval_minutes=5, logger=None)
.enable_auto_save() / .disable_auto_save()
.mark_data_modified() / .force_save()
Legacy test API (expected by tests/test_auto_save.py):
AutoSaveManager(data_file_path=..., backup_dir=..., status_callback=...,
error_callback=..., interval_minutes=0.1, max_backups=3)
.start() / .stop()
.create_backup(suffix) / .get_backup_files() / .restore_from_backup(path)
Both modes share a single implementation for simplicity. Mode is inferred by
presence of 'data_file_path' in kwargs (legacy) vs 'save_callback' (new).
"""
from __future__ import annotations
import contextlib
import glob
import os
import re
import shutil
import threading
from collections.abc import Callable
from datetime import datetime
from typing import Any
from constants import BACKUP_PATH
class AutoSaveManager:
"""Manages automatic saving of user data at regular intervals."""
"""Unified auto-save & backup manager supporting legacy and new APIs."""
def __init__(
self, save_callback: Callable[[], None], interval_minutes: int = 5, logger=None
) -> None:
"""
Initialize auto-save manager.
# ------------------------------------------------------------------
# Construction / mode detection
# ------------------------------------------------------------------
def __init__(self, *args, **kwargs) -> None: # type: ignore[override]
# Determine mode: legacy if a filesystem path is provided
self._legacy_mode = "data_file_path" in kwargs or (
args and isinstance(args[0], str)
)
self.logger = kwargs.get("logger")
Args:
save_callback: Function to call for saving data
interval_minutes: Minutes between auto-saves (default: 5)
logger: Logger instance for debugging
"""
self.save_callback = save_callback
self.interval_seconds = interval_minutes * 60
self.logger = logger
self._auto_save_enabled = False
self._save_thread: threading.Thread | None = None
self._stop_event = threading.Event()
self._last_save_time: datetime | None = None
self._data_modified = False
if self._legacy_mode:
# Legacy parameters (tests expect these attributes)
self.data_file_path: str = kwargs.get(
"data_file_path", args[0] if args else ""
)
self.backup_dir: str = kwargs.get("backup_dir", BACKUP_PATH)
self.status_callback: Callable[[str], None] | None = kwargs.get(
"status_callback"
)
self.error_callback: Callable[[str], None] | None = kwargs.get(
"error_callback"
)
self.interval_minutes: float = float(kwargs.get("interval_minutes", 5))
self.max_backups: int = int(kwargs.get("max_backups", 10))
self.interval_seconds: float = self.interval_minutes * 60
self.save_callback: Callable[[], None] | None = None # Not used in tests
self._thread: threading.Thread | None = None
self._stop_event = threading.Event()
self.is_running: bool = False
self._last_save_time: datetime | None = None
self._data_modified = False # Unused in legacy tests but kept
self._ensure_backup_directory()
else:
# New application mode
save_cb: Callable[[], None] | None = kwargs.get("save_callback")
if save_cb is None and args:
save_cb = args[0]
interval = float(kwargs.get("interval_minutes", 5))
self.save_callback = save_cb
self.interval_minutes = interval
self.interval_seconds = interval * 60
self._auto_save_enabled = False
self._save_thread: threading.Thread | None = None
self._stop_event = threading.Event()
self._last_save_time: datetime | None = None
self._data_modified = False
# Shim attributes for compatibility (unused in new mode)
self.data_file_path = ""
self.backup_dir = BACKUP_PATH
self.status_callback = None
self.error_callback = None
self.max_backups = 10
self.is_running = False
def enable_auto_save(self) -> None:
"""Enable automatic saving."""
if self._auto_save_enabled:
if self._legacy_mode:
# Map to legacy start()
self.start()
return
if getattr(self, "_auto_save_enabled", False):
return
self._auto_save_enabled = True
self._stop_event.clear()
self._save_thread = threading.Thread(target=self._auto_save_loop, daemon=True)
self._save_thread.start()
if self.logger:
interval_minutes = self.interval_seconds / 60
self.logger.info(
f"Auto-save enabled with {interval_minutes:.1f} minute intervals"
f"Auto-save enabled with {self.interval_minutes:.1f} minute intervals"
)
def disable_auto_save(self) -> None:
"""Disable automatic saving."""
if not self._auto_save_enabled:
if self._legacy_mode:
self.stop()
return
if not getattr(self, "_auto_save_enabled", False):
return
self._auto_save_enabled = False
self._stop_event.set()
if self._save_thread and self._save_thread.is_alive():
self._save_thread.join(timeout=2.0)
if self.logger:
self.logger.info("Auto-save disabled")
@@ -67,15 +125,14 @@ class AutoSaveManager:
def force_save(self) -> None:
"""Force an immediate save if data has been modified."""
if self._data_modified:
if self._data_modified and self.save_callback:
try:
self.save_callback()
self._last_save_time = datetime.now()
self._data_modified = False
if self.logger:
self.logger.debug("Force save completed successfully")
except Exception as e:
except Exception as e: # pragma: no cover - defensive
if self.logger:
self.logger.error(f"Force save failed: {e}")
@@ -85,7 +142,11 @@ class AutoSaveManager:
def is_enabled(self) -> bool:
"""Check if auto-save is currently enabled."""
return self._auto_save_enabled
return (
self.is_running
if self._legacy_mode
else getattr(self, "_auto_save_enabled", False)
)
def has_unsaved_changes(self) -> bool:
"""Check if there are unsaved changes."""
@@ -94,16 +155,14 @@ class AutoSaveManager:
def _auto_save_loop(self) -> None:
"""Main auto-save loop running in background thread."""
while not self._stop_event.wait(self.interval_seconds):
if self._data_modified:
if self._data_modified and self.save_callback:
try:
self.save_callback()
self._last_save_time = datetime.now()
self._data_modified = False
if self.logger:
self.logger.debug("Auto-save completed successfully")
except Exception as e:
except Exception as e: # pragma: no cover - defensive
if self.logger:
self.logger.error(f"Auto-save failed: {e}")
@@ -116,212 +175,195 @@ class AutoSaveManager:
"""
if not 1 <= minutes <= 60:
raise ValueError("Auto-save interval must be between 1 and 60 minutes")
old_interval = self.interval_seconds / 60
self.interval_seconds = minutes * 60
old = self.interval_minutes
self.interval_minutes = float(minutes)
self.interval_seconds = self.interval_minutes * 60
if self.logger:
self.logger.info(
f"Auto-save interval changed from {old_interval:.1f} "
f"to {minutes} minutes"
"Auto-save interval changed from %.1f to %.1f minutes",
old,
self.interval_minutes,
)
# Restart auto-save with new interval if it was running
if self._auto_save_enabled:
if not self._legacy_mode and getattr(self, "_auto_save_enabled", False):
self.disable_auto_save()
self.enable_auto_save()
def cleanup(self) -> None:
"""Clean up resources when shutting down."""
self.disable_auto_save()
# Perform final save if there are unsaved changes
if self._legacy_mode:
self.stop()
else:
self.disable_auto_save()
if self._data_modified:
if self.logger:
self.logger.info("Performing final save on cleanup")
self.force_save()
# ------------------------------------------------------------------
# Legacy mode API (periodic file backups)
# ------------------------------------------------------------------
def start(self) -> None:
if not self._legacy_mode or self.is_running:
return
self.is_running = True
self._stop_event.clear()
with contextlib.suppress(Exception):
self.create_backup("startup")
def _loop() -> None:
while not self._stop_event.wait(self.interval_seconds):
with contextlib.suppress(Exception):
self.create_backup("auto")
self._thread = threading.Thread(target=_loop, daemon=True)
self._thread.start()
def stop(self) -> None:
if not self._legacy_mode or not self.is_running:
return
self.is_running = False
self._stop_event.set()
if self._thread and self._thread.is_alive():
self._thread.join(timeout=2.0)
# --------------------- Backup helpers (legacy) ---------------------
def _ensure_backup_directory(self) -> None:
os.makedirs(self.backup_dir, exist_ok=True)
def create_backup(self, suffix: str) -> str | None:
if not getattr(self, "data_file_path", ""):
return None
if not os.path.exists(self.data_file_path):
if self.error_callback:
self.error_callback("Source file does not exist")
return None
safe_suffix = re.sub(r"[^A-Za-z0-9_\-]+", "_", suffix.strip()) or "backup"
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
base = os.path.splitext(os.path.basename(self.data_file_path))[0]
filename = f"{base}_{safe_suffix}_{timestamp}.csv"
dest = os.path.join(self.backup_dir, filename)
try:
shutil.copy2(self.data_file_path, dest)
if self.status_callback:
self.status_callback(f"Backup created: {dest}")
self._cleanup_old_backups()
return dest
except Exception as e: # pragma: no cover - defensive
if self.error_callback:
self.error_callback(f"Backup failed: {e}")
return None
def _cleanup_old_backups(self) -> None:
pattern = os.path.join(self.backup_dir, "*.csv")
files = glob.glob(pattern)
if len(files) <= self.max_backups:
return
files.sort(key=os.path.getmtime, reverse=True)
for f in files[self.max_backups :]:
with contextlib.suppress(Exception):
os.remove(f)
def get_backup_files(self) -> list[str]:
pattern = os.path.join(self.backup_dir, "*.csv")
files = glob.glob(pattern)
files.sort(key=os.path.getmtime, reverse=True)
return files
def restore_from_backup(self, backup_path: str) -> bool:
if not os.path.exists(backup_path):
if self.error_callback:
self.error_callback("Backup file does not exist")
return False
try:
shutil.copy2(backup_path, self.data_file_path)
if self.status_callback:
self.status_callback(f"Restored from backup: {backup_path}")
return True
except Exception as e: # pragma: no cover
if self.error_callback:
self.error_callback(f"Restore failed: {e}")
return False
class BackupManager:
"""Manages automatic backup creation for data files."""
"""Standalone backup manager used by application code."""
def __init__(
self, data_file_path: str, backup_directory: str = BACKUP_PATH, logger=None
):
"""
Initialize backup manager.
Args:
data_file_path: Path to the main data file
backup_directory: Directory to store backups
logger: Logger instance for debugging
"""
self,
data_file_path: str,
backup_directory: str = BACKUP_PATH,
logger=None,
status_callback: Callable[[str], None] | None = None,
) -> None:
self.data_file_path = data_file_path
self.backup_directory = backup_directory
self.logger = logger
self.status_callback = status_callback
self._ensure_backup_directory()
def _ensure_backup_directory(self) -> None:
"""Create backup directory if it doesn't exist."""
import os
os.makedirs(self.backup_directory, exist_ok=True)
def create_backup(self, backup_type: str = "manual") -> str | None:
"""
Create a backup of the data file.
Args:
backup_type: Type of backup ("manual", "auto", "daily")
Returns:
Path to created backup file, or None if backup failed
"""
import os
import shutil
from datetime import datetime
if not os.path.exists(self.data_file_path):
if self.logger:
self.logger.warning("Cannot create backup: data file doesn't exist")
return None
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
base_name = os.path.splitext(os.path.basename(self.data_file_path))[0]
backup_filename = f"{base_name}_backup_{backup_type}_{timestamp}.csv"
backup_path = os.path.join(self.backup_directory, backup_filename)
shutil.copy2(self.data_file_path, backup_path)
msg = f"Backup created: {backup_path}"
if self.logger:
self.logger.info(f"Backup created: {backup_path}")
self.logger.info(msg)
if self.status_callback:
self.status_callback(msg)
return backup_path
except Exception as e:
except Exception as e: # pragma: no cover - defensive
if self.logger:
self.logger.error(f"Backup creation failed: {e}")
return None
def cleanup_old_backups(self, keep_count: int = 10) -> None:
"""
Remove old backup files, keeping only the most recent ones.
Args:
keep_count: Number of backup files to keep
"""
import glob
import os
try:
backup_pattern = os.path.join(self.backup_directory, "*_backup_*.csv")
backup_files = glob.glob(backup_pattern)
if len(backup_files) <= keep_count:
return
# Sort by modification time (newest first)
backup_files.sort(key=os.path.getmtime, reverse=True)
# Remove old files
files_to_remove = backup_files[keep_count:]
for file_path in files_to_remove:
os.remove(file_path)
if self.logger:
self.logger.debug(f"Removed old backup: {file_path}")
removed = 0
for file_path in backup_files[keep_count:]:
with contextlib.suppress(Exception):
os.remove(file_path)
removed += 1
msg = f"Cleaned up {removed} old backup files"
if self.logger:
self.logger.info(f"Cleaned up {len(files_to_remove)} old backup files")
except Exception as e:
self.logger.info(msg)
if self.status_callback and removed:
self.status_callback(msg)
except Exception as e: # pragma: no cover - defensive
if self.logger:
self.logger.error(f"Backup cleanup failed: {e}")
def restore_from_backup(self, backup_path: str) -> bool:
"""
Restore data from a backup file.
Args:
backup_path: Path to the backup file to restore
Returns:
True if restoration was successful, False otherwise
"""
import os
import shutil
if not os.path.exists(backup_path):
if self.logger:
self.logger.error(f"Backup file doesn't exist: {backup_path}")
return False
try:
# Create a backup of current data before restoring
current_backup = self.create_backup("pre_restore")
# Restore from backup
shutil.copy2(backup_path, self.data_file_path)
msg = f"Successfully restored from backup: {backup_path}"
if self.logger:
self.logger.info(f"Successfully restored from backup: {backup_path}")
self.logger.info(msg)
if current_backup:
self.logger.info(f"Previous data backed up to: {current_backup}")
if self.status_callback:
self.status_callback(msg)
return True
except Exception as e:
except Exception as e: # pragma: no cover - defensive
if self.logger:
self.logger.error(f"Restore from backup failed: {e}")
return False
def list_backups(self) -> list[dict[str, Any]]:
"""
List all available backup files with their details.
Returns:
List of dictionaries containing backup file information
"""
import glob
import os
from datetime import datetime
backup_pattern = os.path.join(self.backup_directory, "*_backup_*.csv")
backup_files = glob.glob(backup_pattern)
backups = []
for backup_path in backup_files:
try:
stat = os.stat(backup_path)
backups.append(
{
"path": backup_path,
"filename": os.path.basename(backup_path),
"size": stat.st_size,
"created": datetime.fromtimestamp(stat.st_mtime),
"type": self._extract_backup_type(backup_path),
}
)
except Exception as e:
if self.logger:
self.logger.warning(f"Error reading backup file {backup_path}: {e}")
# Sort by creation time (newest first)
backups.sort(key=lambda x: x["created"], reverse=True)
return backups
def _extract_backup_type(self, backup_path: str) -> str:
"""Extract backup type from filename."""
import os
filename = os.path.basename(backup_path)
if "_backup_auto_" in filename:
return "auto"
elif "_backup_daily_" in filename:
return "daily"
elif "_backup_manual_" in filename:
return "manual"
elif "_backup_pre_restore_" in filename:
return "pre_restore"
else:
return "unknown"