crayonix/nimbusflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backend): add database versioning

Browse Source
This commit is contained in:
Giovani
2025-08-28 10:04:56 -04:00
parent 4df946731a
commit 954abb704e
6 changed files with 217 additions and 603 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
backend/cli/ascii.txt
View File

@@ -1,42 +0,0 @@
████████████████████████████████████████████████████████████████████████████████████████████████████
████████████████████████████████████████████████████████████████████████████████████████████████████
████████████████████████████████████████████████████████████████████████████████████████████████████
█████████████████████████████████████████████▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░▒▒▒▒▒▓▓████████████████████████████
███████████████████████████████████▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░░▒▒▒████████████████████
████████████████████████████▓▒▒▒▒▒▒▒▒▓▓████████████████████████████████▓▒▒▒▒▒▒░░░░░▒▓███████████████
██████████████████████▓▒▒▒▒▓██████████████████████████████████████████████████▒▒▒▒░░░░░▒████████████
█████████████████▓▒▒▓████████████████████████████████████████████████████████████▓▒▒▒░░░░▒██████████
█████████████▓▓▓████████████████████████████████████████████████████████████████████▒▒▒░░░░▒████████
██████████▓██████████████████████████████████████████████████████████████████████████▓▒▒░░░░▒███████
███████████████████████████████████████▓▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████▓▒░░░░░▒██████
████████████████████████▓▓▓▓█████████▓▒▒▒▒▒░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒█████████████████████████▒▒░░░░▒▓█████
███████████████████▓▒▒▒▒░░░░░░▒▓▓▒▒▒▒▒▒▒▒░░░░░░░░░░░░░░░░░░░░░▒▒██████████████████████▓▒░░░░░▒▒█████
█████████████████▓▒▒▒░░░░░░░░░░░░▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒█████████████████████▒░░░░░▒▒▓█████
████████████████▒▒▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒███████████████████▒░░░░░▒▒▒▓█████
███████████████▓▒▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒████████████████▓▒░░░░░░▒▒▒▒██████
███████████████▒▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒▒██████████████▒░░░░░░░▒▒▒▒▒███████
█████████████▓▒▒▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒░░░░░░░░░░░░▒▒▒▒▒▓███████▓▒░░░░░░░░░▒▒▒▒▒▓████████
███████████▒▒▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒▒░░░░░░░░░░░░▒▒▒▒▓▓▒░░░░░░░░░░░░▒▒▒▒▒▒██████████
██████████▓▒░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░▒▒▒░░░░░░░░░░░░▒▒▒▒░░░░░░░░░░▒▒▒▒▒▒▒▒▓███████████
█████████▒▒▒▒░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒░░░░░░░░░░░░░░░░▒▒▒▒░░░░░░░░░░░▒▒▒▒░░░░░░▒▒▒▒▒▒▒▒▒▓██████████████
███████▒▒▒▒░░░░░░░░░░▒▒▒▒▒░░░░░░░░▒▒▒░░░░░░░░░░░░░░▒▒▒▒░░░░░▒░░░░▒▒▒▒▒░▒▒▒▒▒▒▒▒▒▒▒▒▓████████████████
██████▒▒▒▒░░░░░░░░░░▒▒▒░░░░░░░░░░░░░░▒▒░░░░░░░░░░░▒▒▒▒░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▓████████████████████
█████▒▒▒▒░░░░░░░░░▒▒▒░░░░░░░░░░░░░░░░░▒▒░░░░░░░░░▒▒▒▒▒░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▓████████████████████████
█████▒▒▒▒░░░░░▒▒░▒▒▒▒░░░░░░░▒▒░░░░░░░░▒▒░░░░░░░░░░▒▒▒▒░░░░▒▒▒▒▒▒▒▒▒▒▒▒▓▓████████████████████████████
█████▒▒▒▒░░░░░▒▒░▒▒▒▒░░░░░▒▒░░░░░░░░░░▒▒▒░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒██████████████████████████████
█████▒▒▒▒░░░░▒▒▒░▒▒▒▒░░░░▒▒░░░░░░░░░░░▒▒░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▓██████████████████████████████
█████▓▒▒▒▒▒░░▒▒▒▒▒▒▒▒░░░░▒▒░░░░░░░░░░▒▒▒░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████
██████▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░▒▒▒▒░░░░░░▒▒▒▒▒▒▒░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▓████████████████████████████████
████████▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▓██████████████████████████████████
██████████▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████████
███████████▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████████
███████████▓▒▒▒▒▒▒▒▒▒▒▒░░░░░░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒████████████████████████████████████████
█████████████▓▒▒▒▒▒▒▒▒▒░░░░░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒█████████████████████████████████████████
███████████████████▓▒▒▒▒░░░░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▓██████████████████████████████████████████
████████████████████▓▒▒▒▒░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▓▒▒▒▒▒▒▒▓█████████████████████████████████████████████
█████████████████████▓▒▒▒▒▒▒░░░▒▒▒▒▒▒▒▒▒▒▒▓█████████████████████████████████████████████████████████
███████████████████████▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒██████████████████████████████████████████████████████████████
███████████████████████████▓▒▒▒▒▒▒▓█████████████████████████████████████████████████████████████████
████████████████████████████████████████████████████████████████████████████████████████████████████
████████████████████████████████████████████████████████████████████████████████████████████████████
████████████████████████████████████████████████████████████████████████████████████████████████████

116
backend/cli/base.py
View File

@@ -2,7 +2,10 @@
Base CLI class and utilities for NimbusFlow CLI.
"""
import shutil
from datetime import datetime
from pathlib import Path
from typing import Optional
from backend.db.connection import DatabaseConnection
from backend.repositories import (
MemberRepository,
@@ -14,6 +17,19 @@ from backend.repositories import (
)
from backend.services.scheduling_service import SchedulingService
# Import Colors from interactive module for consistent styling
try:
from .interactive import Colors
except ImportError:
# Fallback colors if interactive module not available
class Colors:
RESET = '\033[0m'
SUCCESS = '\033[1m\033[92m'
WARNING = '\033[1m\033[93m'
ERROR = '\033[1m\033[91m'
CYAN = '\033[96m'
DIM = '\033[2m'
class CLIError(Exception):
"""Custom exception for CLI-specific errors."""
@@ -21,17 +37,105 @@ class CLIError(Exception):
class NimbusFlowCLI:
"""Main CLI application class."""
"""Main CLI application class with database versioning."""
def __init__(self, db_path: str = "database6_accepts_and_declines.db"):
"""Initialize CLI with database connection."""
self.db_path = Path(__file__).parent.parent / db_path
if not self.db_path.exists():
raise CLIError(f"Database not found: {self.db_path}")
def __init__(self, db_path: str = "database.db", create_version: bool = True):
"""Initialize CLI with database connection, always using most recent version."""
self.db_dir = Path(__file__).parent.parent / "db" / "sqlite"
self.base_db_path = self.db_dir / db_path
# Always find and use the most recent database version
self.db_path = self._get_most_recent_database()
if create_version:
# Create a new version based on the most recent one
self.db_path = self._create_versioned_database()
self.db = DatabaseConnection(self.db_path)
self._init_repositories()
def _get_most_recent_database(self) -> Path:
"""Get the most recent database version, or base database if no versions exist."""
versions = self.list_database_versions()
if versions:
# Return the most recent versioned database
most_recent = versions[0] # Already sorted newest first
return most_recent
else:
# No versions exist, use base database
if not self.base_db_path.exists():
raise CLIError(f"Base database not found: {self.base_db_path}")
return self.base_db_path
def _create_versioned_database(self) -> Path:
"""Create a versioned copy from the most recent database."""
source_db = self.db_path # Use the most recent database as source
# Generate timestamp-based version
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
version = self._get_next_version_number()
versioned_name = f"database_v{version}_{timestamp}.db"
versioned_path = self.db_dir / versioned_name
# Copy the most recent database to create the versioned copy
shutil.copy2(source_db, versioned_path)
print(f"{Colors.SUCCESS}Created versioned database:{Colors.RESET} {Colors.CYAN}{versioned_name}{Colors.RESET}")
print(f"{Colors.DIM}Based on: {source_db.name}{Colors.RESET}")
return versioned_path
def _get_next_version_number(self) -> int:
"""Get the next version number by checking existing versioned databases."""
version_pattern = "database_v*_*.db"
existing_versions = list(self.db_dir.glob(version_pattern))
if not existing_versions:
return 1
# Extract version numbers from existing files
versions = []
for db_file in existing_versions:
try:
# Parse version from filename like "database_v123_20250828_143022.db"
parts = db_file.stem.split('_')
if len(parts) >= 2 and parts[1].startswith('v'):
version_num = int(parts[1][1:]) # Remove 'v' prefix
versions.append(version_num)
except (ValueError, IndexError):
continue
return max(versions) + 1 if versions else 1
def list_database_versions(self) -> list[Path]:
"""List all versioned databases in chronological order."""
version_pattern = "database_v*_*.db"
versioned_dbs = list(self.db_dir.glob(version_pattern))
# Sort by modification time (newest first)
return sorted(versioned_dbs, key=lambda x: x.stat().st_mtime, reverse=True)
def cleanup_old_versions(self, keep_latest: int = 5) -> int:
"""Clean up old database versions, keeping only the latest N versions."""
versions = self.list_database_versions()
if len(versions) <= keep_latest:
return 0
versions_to_delete = versions[keep_latest:]
deleted_count = 0
for db_path in versions_to_delete:
try:
db_path.unlink()
deleted_count += 1
print(f"{Colors.DIM}Deleted old version: {db_path.name}{Colors.RESET}")
except OSError as e:
print(f"{Colors.WARNING}⚠️ Could not delete {db_path.name}: {e}{Colors.RESET}")
return deleted_count
def _init_repositories(self):
"""Initialize all repository instances."""
self.member_repo = MemberRepository(self.db)

94
backend/cli/interactive.py
View File

@@ -3,6 +3,7 @@ Interactive CLI interface for NimbusFlow.
"""
import sys
import time
from pathlib import Path
from typing import TYPE_CHECKING
@@ -35,6 +36,13 @@ class Colors:
ERROR = '\033[1m\033[91m' # Bold Red
WARNING = '\033[1m\033[93m' # Bold Yellow
INPUT_BOX = '\033[90m' # Grey
# Gold shimmer colors
GOLD_DARK = '\033[38;5;130m' # Dark gold
GOLD_MEDIUM = '\033[38;5;178m' # Medium gold
GOLD_BRIGHT = '\033[38;5;220m' # Bright gold
GOLD_SHINE = '\033[1m\033[38;5;226m' # Bright shining gold
GOLD_WHITE = '\033[1m\033[97m' # Bright white for peak shine
def create_input_box(prompt: str, width: int = 60) -> str:
@@ -61,27 +69,77 @@ def clear_screen():
print("\033[2J\033[H")
def get_shimmer_color(position: int, shimmer_center: int, shimmer_width: int = 8) -> str:
"""Get the appropriate shimmer color based on distance from shimmer center."""
distance = abs(position - shimmer_center)
if distance == 0:
return Colors.GOLD_WHITE
elif distance == 1:
return Colors.GOLD_SHINE
elif distance <= 3:
return Colors.GOLD_BRIGHT
elif distance <= 5:
return Colors.GOLD_MEDIUM
elif distance <= shimmer_width:
return Colors.GOLD_DARK
else:
return Colors.GOLD_DARK
def animate_nimbusflow_text() -> None:
"""Animate the NimbusFlow ASCII text and frame with a gold shimmer effect."""
# Complete welcome screen lines including borders
welcome_lines = [
"╔════════════════════════════════════════════════════════════════════════════════════════════╗",
"║ ║",
"║ ███╗ ██╗██╗███╗ ███╗██████╗ ██╗ ██╗███████╗ ███████╗██╗ ██████╗ ██╗ ██╗ ║",
"║ ████╗ ██║██║████╗ ████║██╔══██╗██║ ██║██╔════╝ ██╔════╝██║ ██╔═══██╗██║ ██║ ║",
"║ ██╔██╗ ██║██║██╔████╔██║██████╔╝██║ ██║███████╗ █████╗ ██║ ██║ ██║██║ █╗ ██║ ║",
"║ ██║╚██╗██║██║██║╚██╔╝██║██╔══██╗██║ ██║╚════██║ ██╔══╝ ██║ ██║ ██║██║███╗██║ ║",
"║ ██║ ╚████║██║██║ ╚═╝ ██║██████╔╝╚██████╔╝███████║ ██║ ███████╗╚██████╔╝╚███╔███╔╝ ║",
"║ ╚═╝ ╚═══╝╚═╝╚═╝ ╚═╝╚═════╝ ╚═════╝ ╚══════╝ ╚═╝ ╚══════╝ ╚═════╝ ╚══╝╚══╝ ║",
"║ ║",
"╚════════════════════════════════════════════════════════════════════════════════════════════╝"
]
# Calculate max width for animation
max_width = max(len(line) for line in welcome_lines)
# Animation parameters
shimmer_width = 12
total_steps = max_width + shimmer_width * 2
step_delay = 0.025 # Seconds between frames (even faster animation)
# Animate the shimmer effect
for step in range(total_steps):
shimmer_center = step - shimmer_width
# Move cursor up to overwrite previous frame (10 lines total)
if step > 0:
print(f"\033[{len(welcome_lines)}A", end="")
for line in welcome_lines:
for i, char in enumerate(line):
if char.isspace():
print(char, end="")
else:
color = get_shimmer_color(i, shimmer_center, shimmer_width)
print(f"{color}{char}{Colors.RESET}", end="")
print() # New line after each row
# Add a small delay for animation
time.sleep(step_delay)
def display_welcome():
"""Display welcome screen."""
"""Display welcome screen with animated shimmer effect."""
print("\033[2J\033[H") # Clear screen and move cursor to top
# NimbusFlow branding
welcome_text = """
╔════════════════════════════════════════════════════════════════════════════════════════════╗
║ ║
║ ███╗ ██╗██╗███╗ ███╗██████╗ ██╗ ██╗███████╗ ███████╗██╗ ██████╗ ██╗ ██╗ ║
║ ████╗ ██║██║████╗ ████║██╔══██╗██║ ██║██╔════╝ ██╔════╝██║ ██╔═══██╗██║ ██║ ║
║ ██╔██╗ ██║██║██╔████╔██║██████╔╝██║ ██║███████╗ █████╗ ██║ ██║ ██║██║ █╗ ██║ ║
║ ██║╚██╗██║██║██║╚██╔╝██║██╔══██╗██║ ██║╚════██║ ██╔══╝ ██║ ██║ ██║██║███╗██║ ║
║ ██║ ╚████║██║██║ ╚═╝ ██║██████╔╝╚██████╔╝███████║ ██║ ███████╗╚██████╔╝╚███╔███╔╝ ║
║ ╚═╝ ╚═══╝╚═╝╚═╝ ╚═╝╚═════╝ ╚═════╝ ╚══════╝ ╚═╝ ╚══════╝ ╚═════╝ ╚══╝╚══╝ ║
║ ║
║ 🎵 Scheduling System 🎵 ║
╚════════════════════════════════════════════════════════════════════════════════════════════╝
"""
print(welcome_text)
print() # Add some top padding
animate_nimbusflow_text()
print()
@@ -355,7 +413,7 @@ def run_interactive_mode(cli: "NimbusFlowCLI"):
display_welcome()
print(f"{Colors.HEADER}Welcome to the NimbusFlow Interactive CLI{Colors.RESET}")
print(f"{Colors.DIM}Navigate through menus to manage your choir scheduling system.{Colors.RESET}")
print(f"{Colors.DIM}Navigate through menus to manage your scheduling system.{Colors.RESET}")
print()
input(f"{Colors.DIM}Press Enter to continue...{Colors.RESET}")

42
backend/cli/main.py
View File

@@ -44,16 +44,30 @@ def main():
if not args.command:
# Launch interactive mode when no command is provided
try:
cli = NimbusFlowCLI()
cli = NimbusFlowCLI(create_version=True) # Always create versioned DB for interactive mode
# Show versioning info with colors
from .base import Colors
versions = cli.list_database_versions()
if len(versions) > 1:
print(f"{Colors.CYAN}Database versions available:{Colors.RESET} {Colors.SUCCESS}{len(versions)}{Colors.RESET}")
print(f"{Colors.CYAN}Using:{Colors.RESET} {Colors.CYAN}{cli.db_path.name}{Colors.RESET}")
# Auto-cleanup if too many versions
if len(versions) > 10:
deleted = cli.cleanup_old_versions(keep_latest=5)
if deleted > 0:
print(f"{Colors.WARNING}Cleaned up {deleted} old database versions{Colors.RESET}")
run_interactive_mode(cli)
except CLIError as e:
print(f"❌ Error: {e}")
print(f"{Colors.ERROR}❌ Error: {e}{Colors.RESET}")
return 1
except KeyboardInterrupt:
print("\n🛑 Interrupted by user")
print(f"\n{Colors.WARNING}🛑 Interrupted by user{Colors.RESET}")
return 1
except Exception as e:
print(f"❌ Unexpected error: {e}")
print(f"{Colors.ERROR}❌ Unexpected error: {e}{Colors.RESET}")
return 1
finally:
if 'cli' in locals():
@@ -61,7 +75,11 @@ def main():
return
try:
cli = NimbusFlowCLI()
cli = NimbusFlowCLI(create_version=False) # Don't version for regular CLI commands
# Show which database is being used for regular commands
from .base import Colors
print(f"{Colors.CYAN}Using database:{Colors.RESET} {Colors.CYAN}{cli.db_path.name}{Colors.RESET}")
# Route commands to their respective handlers
if args.command == "members":
@@ -70,7 +88,7 @@ def main():
elif args.members_action == "show":
cmd_members_show(cli, args)
else:
print("❌ Unknown members action. Use 'list' or 'show'")
print(f"{Colors.ERROR}❌ Unknown members action. Use 'list' or 'show'{Colors.RESET}")
elif args.command == "schedules":
if args.schedules_action == "list":
@@ -84,25 +102,25 @@ def main():
elif args.schedules_action == "schedule":
cmd_schedules_schedule(cli, args)
else:
print("❌ Unknown schedules action. Use 'list', 'show', 'accept', 'decline', or 'schedule'")
print(f"{Colors.ERROR}❌ Unknown schedules action. Use 'list', 'show', 'accept', 'decline', or 'schedule'{Colors.RESET}")
elif args.command == "services":
if args.services_action == "list":
cmd_services_list(cli, args)
else:
print("❌ Unknown services action. Use 'list'")
print(f"{Colors.ERROR}❌ Unknown services action. Use 'list'{Colors.RESET}")
else:
print(f"❌ Unknown command: {args.command}")
print(f"{Colors.ERROR}❌ Unknown command: {args.command}{Colors.RESET}")
except CLIError as e:
print(f"❌ Error: {e}")
print(f"{Colors.ERROR}❌ Error: {e}{Colors.RESET}")
return 1
except KeyboardInterrupt:
print("\n🛑 Interrupted by user")
print(f"\n{Colors.WARNING}🛑 Interrupted by user{Colors.RESET}")
return 1
except Exception as e:
print(f"❌ Unexpected error: {e}")
print(f"{Colors.ERROR}❌ Unexpected error: {e}{Colors.RESET}")
return 1
finally:
if 'cli' in locals():

524
backend/database/repository.py
View File

@@ -1,524 +0,0 @@
import datetime as dt
from typing import Optional, Tuple, List
from backend.db.connection import DatabaseConnection
from backend.models import (
Classification,
Member,
ServiceType,
Service,
ServiceAvailability,
Schedule,
AcceptedLog,
DeclineLog,
ScheduledLog,
)
class Repository:
"""
High‑level data‑access layer.
Responsibilities
----------------
* CRUD helpers for the core tables.
* Round‑robin queue that respects:
- Members.LastAcceptedAt (fair order)
- Members.LastDeclinedAt (one‑day cool‑off)
* “Reservation” handling using the **Schedules** table
(pending → accepted → declined).
* Audit logging (AcceptedLog, DeclineLog, ScheduledLog).
"""
def __init__(self, db: DatabaseConnection):
self.db = db
# -----------------------------------------------------------------
# CRUD helpers – they now return model objects (or IDs)
# -----------------------------------------------------------------
# -----------------------------------------------------------------
# CREATE
# -----------------------------------------------------------------
def create_classification(self, classification_name: str) -> Classification:
"""Insert a new classification and return the saved model."""
classification = Classification(
ClassificationId=-1, # placeholder – will be replaced by DB
ClassificationName=classification_name,
)
# Build INSERT statement from the dataclass dict (skip PK)
data = classification.to_dict()
data.pop("ClassificationId") # AUTOINCREMENT column
cols = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
sql = f"INSERT INTO Classifications ({cols}) VALUES ({placeholders})"
self.db.execute(sql, tuple(data.values()))
classification.ClassificationId = self.db.lastrowid
return classification
def create_member(
self,
first_name: str,
last_name: str,
email: Optional[str] = None,
phone_number: Optional[str] = None,
classification_id: Optional[int] = None,
notes: Optional[str] = None,
is_active: int = 1,
) -> Member:
"""Insert a new member and return the saved model."""
member = Member(
MemberId=-1,
FirstName=first_name,
LastName=last_name,
Email=email,
PhoneNumber=phone_number,
ClassificationId=classification_id,
Notes=notes,
IsActive=is_active,
LastAcceptedAt=None,
LastDeclinedAt=None,
)
data = member.to_dict()
data.pop("MemberId") # let SQLite fill the PK
cols = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
sql = f"INSERT INTO Members ({cols}) VALUES ({placeholders})"
self.db.execute(sql, tuple(data.values()))
member.MemberId = self.db.lastrowid
return member
def create_service_type(self, type_name: str) -> ServiceType:
"""Insert a new service type."""
st = ServiceType(ServiceTypeId=-1, TypeName=type_name)
data = st.to_dict()
data.pop("ServiceTypeId")
cols = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
sql = f"INSERT INTO ServiceTypes ({cols}) VALUES ({placeholders})"
self.db.execute(sql, tuple(data.values()))
st.ServiceTypeId = self.db.lastrowid
return st
def create_service(self, service_type_id: int, service_date: dt.date) -> Service:
"""Insert a new service row (date + type)."""
sv = Service(ServiceId=-1, ServiceTypeId=service_type_id, ServiceDate=service_date)
data = sv.to_dict()
data.pop("ServiceId")
cols = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
sql = f"INSERT INTO Services ({cols}) VALUES ({placeholders})"
self.db.execute(sql, tuple(data.values()))
sv.ServiceId = self.db.lastrowid
return sv
def create_service_availability(self, member_id: int, service_type_id: int) -> ServiceAvailability:
"""Link a member to a service type (availability matrix)."""
sa = ServiceAvailability(
ServiceAvailabilityId=-1,
MemberId=member_id,
ServiceTypeId=service_type_id,
)
data = sa.to_dict()
data.pop("ServiceAvailabilityId")
cols = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data)
sql = f"INSERT INTO ServiceAvailability ({cols}) VALUES ({placeholders})"
self.db.execute(sql, tuple(data.values()))
sa.ServiceAvailabilityId = self.db.lastrowid
return sa
# -----------------------------------------------------------------
# READ – return **lists of models**
# -----------------------------------------------------------------
def get_all_classifications(self) -> List[Classification]:
rows = self.db.fetchall("SELECT * FROM Classifications")
return [Classification.from_row(r) for r in rows]
def get_all_members(self) -> List[Member]:
rows = self.db.fetchall("SELECT * FROM Members")
return [Member.from_row(r) for r in rows]
def get_all_service_types(self) -> List[ServiceType]:
rows = self.db.fetchall("SELECT * FROM ServiceTypes")
return [ServiceType.from_row(r) for r in rows]
def get_all_services(self) -> List[Service]:
rows = self.db.fetchall("SELECT * FROM Services")
return [Service.from_row(r) for r in rows]
def get_all_service_availability(self) -> List[ServiceAvailability]:
rows = self.db.fetchall("SELECT * FROM ServiceAvailability")
return [ServiceAvailability.from_row(r) for r in rows]
# -----------------------------------------------------------------
# INTERNAL helpers used by the queue logic
# -----------------------------------------------------------------
def _lookup_classification(self, name: str) -> int:
"""Return ClassificationId for a given name; raise if missing."""
row = self.db.fetchone(
"SELECT ClassificationId FROM Classifications WHERE ClassificationName = ?",
(name,),
)
if row is None:
raise ValueError(f'Classification "{name}" does not exist')
return row["ClassificationId"]
def _ensure_service(self, service_date: dt.date) -> int:
"""
Return a ServiceId for ``service_date``.
If the row does not exist we create a generic Service row
(using the first ServiceType as a default).
"""
row = self.db.fetchone(
"SELECT ServiceId FROM Services WHERE ServiceDate = ?", (service_date,)
)
if row:
return row["ServiceId"]
default_type = self.db.fetchone(
"SELECT ServiceTypeId FROM ServiceTypes LIMIT 1"
)
if not default_type:
raise RuntimeError(
"No ServiceTypes defined – cannot create a Service row"
)
self.db.execute(
"INSERT INTO Services (ServiceTypeId, ServiceDate) VALUES (?,?)",
(default_type["ServiceTypeId"], service_date),
)
return self.db.lastrowid
def has_schedule_for_service(
self,
member_id: int,
service_id: int,
status: str,
include_expired: bool = False,
) -> bool:
"""
Return True if the member has a schedule row for the given ``service_id``
with the specified ``status``.
For ``status='pending'`` the default behaviour is to ignore rows whose
``ExpiresAt`` timestamp is already in the past (they are not actionable).
Set ``include_expired=True`` if you deliberately want to see *any* pending
row regardless of its expiration.
Parameters
----------
member_id : int
The member we are inspecting.
service_id : int
The service we are interested in.
status : str
One of the schedule statuses (e.g. ``'accepted'`` or ``'pending'``).
include_expired : bool, optional
When checking for pending rows, ignore the expiration guard if set to
``True``. Defaults to ``False`` (i.e. only non‑expired pending rows
count).
Returns
-------
bool
True if a matching row exists, otherwise False.
"""
sql = """
SELECT 1
FROM Schedules
WHERE MemberId = ?
AND ServiceId = ?
AND Status = ?
"""
args = [member_id, service_id, status]
# Guard against expired pending rows unless the caller explicitly wants them.
if not include_expired and status == "pending":
sql += " AND ExpiresAt > CURRENT_TIMESTAMP"
sql += " LIMIT 1"
row = self.db.fetchone(sql, tuple(args))
return row is not None
def schedule_next_member(
self,
classification_id: int,
service_id: int,
only_active: bool = True,
) -> Optional[Tuple[int, str, str, int]]:
"""
Choose the next member for ``service_id`` while respecting ServiceAvailability.
Ordering (high‑level):
1️⃣ 5‑day decline boost – only if DeclineStreak < 2.
2️⃣ Oldest LastAcceptedAt (round‑robin).
3️⃣ Oldest LastScheduledAt (tie‑breaker).
Skipped if any of the following is true:
• Member lacks a ServiceAvailability row for the ServiceType of ``service_id``.
• Member already has an *accepted* schedule for this service.
• Member already has a *pending* schedule for this service.
• Member already has a *declined* schedule for this service.
"""
# -----------------------------------------------------------------
# 0️⃣ Resolve ServiceTypeId (and ServiceDate) from the Services table.
# -----------------------------------------------------------------
svc_row = self.db.fetchone(
"SELECT ServiceTypeId, ServiceDate FROM Services WHERE ServiceId = ?",
(service_id,),
)
if not svc_row:
# No such service – nothing to schedule.
return None
service_type_id = svc_row["ServiceTypeId"]
# If you need the actual calendar date later you can use:
# service_date = dt.datetime.strptime(svc_row["ServiceDate"], "%Y-%m-%d").date()
# -----------------------------------------------------------------
# 1️⃣ Pull the candidate queue, ordered per the existing rules.
# -----------------------------------------------------------------
BOOST_SECONDS = 5 * 24 * 60 * 60 # 5 days
now_iso = dt.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
sql = f"""
SELECT
MemberId,
FirstName,
LastName,
LastAcceptedAt,
LastScheduledAt,
LastDeclinedAt,
DeclineStreak
FROM Members
WHERE ClassificationId = ?
{"AND IsActive = 1" if only_active else ""}
ORDER BY
/* ① 5‑day boost (only when streak < 2) */
CASE
WHEN DeclineStreak < 2
AND LastDeclinedAt IS NOT NULL
AND julianday(?) - julianday(LastDeclinedAt) <= (? / 86400.0)
THEN 0 -- boosted to the front
ELSE 1
END,
/* ② Round‑robin: oldest acceptance first */
COALESCE(LastAcceptedAt, '1970-01-01') ASC,
/* ③ Tie‑breaker: oldest offer first */
COALESCE(LastScheduledAt, '1970-01-01') ASC
"""
queue = self.db.fetchall(sql, (classification_id, now_iso, BOOST_SECONDS))
# -----------------------------------------------------------------
# 2️⃣ Walk the ordered queue and apply availability + status constraints.
# -----------------------------------------------------------------
for member in queue:
member_id = member["MemberId"]
# ----- Availability check -------------------------------------------------
# Skip members that do NOT have a row in ServiceAvailability for this
# ServiceType.
avail_ok = self.db.fetchone(
"""
SELECT 1
FROM ServiceAvailability
WHERE MemberId = ?
AND ServiceTypeId = ?
LIMIT 1
""",
(member_id, service_type_id),
)
if not avail_ok:
continue # Not eligible for this service type.
# ----- Status constraints (all by service_id) ----------------------------
# a) Already *accepted* for this service?
if self.has_schedule_for_service(member_id, service_id, status="accepted"):
continue
# b) Existing *pending* reservation for this service?
if self.has_schedule_for_service(member_id, service_id, status="pending"):
continue
# c) Already *declined* this service?
if self.has_schedule_for_service(member_id, service_id, status="declined"):
continue
# -------------------------------------------------------------
# SUCCESS – create a pending schedule (minimal columns).
# -------------------------------------------------------------
self.db.execute(
"""
INSERT INTO Schedules
(ServiceId, MemberId, Status)
VALUES
(?,?,?)
""",
(service_id, member_id, "pending"),
)
schedule_id = self.db.lastrowid
# -------------------------------------------------------------
# Update the member's LastScheduledAt so the round‑robin stays fair.
# -------------------------------------------------------------
self.db.execute(
"""
UPDATE Members
SET LastScheduledAt = CURRENT_TIMESTAMP
WHERE MemberId = ?
""",
(member_id,),
)
# -------------------------------------------------------------
# Audit log – historic record (no ScheduleId column any more).
# -------------------------------------------------------------
self.db.execute(
"""
INSERT INTO ScheduledLog (MemberId, ServiceId)
VALUES (?,?)
""",
(member_id, service_id),
)
# -------------------------------------------------------------
# Return the useful bits to the caller.
# -------------------------------------------------------------
return (
member_id,
member["FirstName"],
member["LastName"],
schedule_id,
)
# -----------------------------------------------------------------
# No eligible member found.
# -----------------------------------------------------------------
return None
# -----------------------------------------------------------------
# ACCEPT / DECLINE workflow (operates on the schedule row)
# -----------------------------------------------------------------
def accept_schedule(self, schedule_id: int) -> None:
"""
Convert a *pending* schedule into a real assignment.
- Updates the schedule row (status → accepted, timestamp).
- Writes an entry into ``AcceptedLog``.
- Updates ``Members.LastAcceptedAt`` (advances round‑robin) and clears any cool‑off.
"""
# Load the pending schedule – raise if it does not exist or is not pending
sched = self.db.fetchone(
"""
SELECT ScheduleId, ServiceId, MemberId
FROM Schedules
WHERE ScheduleId = ?
AND Status = 'pending'
""",
(schedule_id,),
)
if not sched:
raise ValueError("Schedule not found or not pending")
service_id = sched["ServiceId"]
member_id = sched["MemberId"]
# 1️⃣ Mark the schedule as accepted
self.db.execute(
"""
UPDATE Schedules
SET Status = 'accepted',
AcceptedAt = CURRENT_TIMESTAMP,
ExpiresAt = CURRENT_TIMESTAMP -- no longer expires
WHERE ScheduleId = ?
""",
(schedule_id,),
)
# 2️⃣ Audit log
self.db.execute(
"""
INSERT INTO AcceptedLog (MemberId, ServiceId)
VALUES (?,?)
""",
(member_id, service_id),
)
# 3️⃣ Advance round‑robin for the member
self.db.execute(
"""
UPDATE Members
SET LastAcceptedAt = CURRENT_TIMESTAMP,
LastDeclinedAt = NULL -- a successful accept clears any cool‑off
WHERE MemberId = ?
""",
(member_id,),
)
def decline_schedule(
self, schedule_id: int, reason: Optional[str] = None
) -> None:
"""
Record that the member declined the offered slot.
Effects
-------
* Inserts a row into ``DeclineLog`` (with the service day).
* Updates ``Members.LastDeclinedAt`` – this implements the one‑day cool‑off.
* Marks the schedule row as ``declined`` (so it can be offered to someone else).
"""
# Load the pending schedule – raise if not found / not pending
sched = self.db.fetchone(
"""
SELECT ScheduleId, ServiceId, MemberId
FROM Schedules
WHERE ScheduleId = ?
AND Status = 'pending'
""",
(schedule_id,),
)
if not sched:
raise ValueError("Schedule not found or not pending")
service_id = sched["ServiceId"]
member_id = sched["MemberId"]
# Need the service *day* for the one‑day cool‑off
svc = self.db.fetchone(
"SELECT ServiceDate FROM Services WHERE ServiceId = ?", (service_id,)
)
if not svc:
raise RuntimeError("Service row vanished while processing decline")
service_day = svc["ServiceDate"] # stored as TEXT 'YYYY‑MM‑DD'
# 1️⃣ Insert into DeclineLog
self.db.execute(
"""
INSERT INTO DeclineLog (MemberId, ServiceId, DeclineDate, Reason)
VALUES (?,?,?,?)
""",
(member_id, service_id, service_day, reason),
)
# 2️⃣ Update the member's cool‑off day
self.db.execute(
"""
UPDATE Members
SET LastDeclinedAt = ?
WHERE MemberId = ?
""",
(service_day, member_id),
)
# 3️⃣ Mark the schedule row as declined
self.db.execute(
"""
UPDATE Schedules
SET Status = 'declined',
DeclinedAt = CURRENT_TIMESTAMP,
DeclineReason = ?
WHERE ScheduleId = ?
""",
(reason, schedule_id),
)

2
backend/main.py
View File

@@ -169,7 +169,7 @@ if __name__ == "__main__":
from backend.repositories import MemberRepository, ScheduleRepository, ServiceRepository, ServiceAvailabilityRepository
from backend.services.scheduling_service import SchedulingService
DB_PATH = Path(__file__).parent / "database6_accepts_and_declines3.db"
DB_PATH = Path(__file__).parent / "db" / "sqlite" / "database.db"
# Initialise DB connection (adjust DSN as needed)
db = DatabaseConnection(DB_PATH)