import os import datetime as dt import sqlite3 import random from pathlib import Path from typing import List, Dict, Set, Tuple # ---------------------------------------------------------------------- # 1️⃣ Helper that creates the database (runs the schema file) # ---------------------------------------------------------------------- def init_db(db_path: Path) -> None: """ If the DB file does not exist, create it and run the schema.sql script. The schema file lives in backend/database/schema.sql. """ if db_path.exists(): print(f"✅ Database already exists at {db_path}") return print(f"🗂️ Creating new SQLite DB at {db_path}") conn = sqlite3.connect(db_path) cur = conn.cursor() schema_path = Path(__file__).parent / "database" / "schema.sql" if not schema_path.is_file(): raise FileNotFoundError(f"Schema file not found: {schema_path}") with open(schema_path, "r", encoding="utf‑8") as f: sql = f.read() cur.executescript(sql) conn.commit() conn.close() print("✅ Schema executed – database is ready.") def next_n_sundays(n: int) -> list[dt.date]: """Return a list with the next `n` Sundays after today.""" today = dt.date.today() # weekday(): Mon=0 … Sun=6 → we want the offset to the *next* Sunday days_until_sunday = (6 - today.weekday()) % 7 # If today is Sunday, days_until_sunday == 0 → we still want the *next* one days_until_sunday = days_until_sunday or 7 first_sunday = today + dt.timedelta(days=days_until_sunday) # Build the list of n Sundays return [first_sunday + dt.timedelta(weeks=i) for i in range(n)] def seed_db(repo) -> None: """ Populate a tiny data‑set, run the round‑robin queue, accept one schedule, decline another and print audit tables. """ print("\n=== 📦 Seeding reference data ===") # ----- classifications ------------------------------------------------- baritone_cls = repo.create_classification("Baritone") tenor_cls = repo.create_classification("Tenor") alto_cls = repo.create_classification("Alto / Mezzo") soprano_cls = repo.create_classification("Soprano") print(f"""Created classifications → {baritone_cls.ClassificationId}=Baritone, {tenor_cls.ClassificationId}=Tenor, {alto_cls.ClassificationId}=Alto, {soprano_cls.ClassificationId}=Soprano\n""") # ----- members -------------------------------------------------------- members = [ # 1‑5 (Tenor) repo.create_member("John", "Doe", "john.doe@example.com", "+155512340001", tenor_cls.ClassificationId), repo.create_member("Mary", "Smith", "mary.smith@example.com", "+155512340002", tenor_cls.ClassificationId), repo.create_member("David", "Lee", "david.lee@example.com", "+155512340003", tenor_cls.ClassificationId), repo.create_member("Emma", "Clark", "emma.clark@example.com", "+155512340004", tenor_cls.ClassificationId), repo.create_member("Jack", "Taylor", "jack.taylor@example.com", "+155512340005", tenor_cls.ClassificationId), # 6‑10 (Alto / Mezzo) repo.create_member("Alice", "Brown", "alice.brown@example.com", "+155512340006", alto_cls.ClassificationId), repo.create_member("Frank", "Davis", "frank.davis@example.com", "+155512340007", alto_cls.ClassificationId), repo.create_member("Grace", "Miller", "grace.miller@example.com", "+155512340008", alto_cls.ClassificationId), repo.create_member("Henry", "Wilson", "henry.wilson@example.com", "+155512340009", alto_cls.ClassificationId), repo.create_member("Isla", "Anderson", "isla.anderson@example.com", "+155512340010", alto_cls.ClassificationId), # 11‑15 (Soprano) repo.create_member("Bob", "Johnson", "bob.johnson@example.com", "+155512340011", soprano_cls.ClassificationId), repo.create_member("Kara", "Thomas", "kara.thomas@example.com", "+155512340012", soprano_cls.ClassificationId), repo.create_member("Liam", "Jackson", "liam.jackson@example.com", "+155512340013", soprano_cls.ClassificationId), repo.create_member("Mia", "White", "mia.white@example.com", "+155512340014", soprano_cls.ClassificationId), repo.create_member("Noah", "Harris", "noah.harris@example.com", "+155512340015", soprano_cls.ClassificationId), # 16‑20 (Baritone) repo.create_member("Olivia", "Martin", "olivia.martin@example.com", "+155512340016", baritone_cls.ClassificationId), repo.create_member("Paul", "Doe", "paul.doe@example.com", "+155512340017", baritone_cls.ClassificationId), repo.create_member("Quinn", "Smith", "quinn.smith@example.com", "+155512340018", baritone_cls.ClassificationId), repo.create_member("Ruth", "Brown", "ruth.brown@example.com", "+155512340019", baritone_cls.ClassificationId), repo.create_member("Sam", "Lee", "sam.lee@example.com", "+155512340020", baritone_cls.ClassificationId), # 21‑25 (Tenor again) repo.create_member("Tina", "Clark", "tina.clark@example.com", "+155512340021", tenor_cls.ClassificationId), repo.create_member("Umar", "Davis", "umar.davis@example.com", "+155512340022", tenor_cls.ClassificationId), repo.create_member("Vera", "Miller", "vera.miller@example.com", "+155512340023", tenor_cls.ClassificationId), repo.create_member("Walt", "Wilson", "walt.wilson@example.com", "+155512340024", tenor_cls.ClassificationId), repo.create_member("Xena", "Anderson", "xena.anderson@example.com", "+155512340025", tenor_cls.ClassificationId), # 26‑30 (Alto / Mezzo again) repo.create_member("Yara", "Thomas", "yara.thomas@example.com", "+155512340026", alto_cls.ClassificationId), repo.create_member("Zane", "Jackson", "zane.jackson@example.com", "+155512340027", alto_cls.ClassificationId), repo.create_member("Anna", "White", "anna.white@example.com", "+155512340028", alto_cls.ClassificationId), repo.create_member("Ben", "Harris", "ben.harris@example.com", "+155512340029", alto_cls.ClassificationId), repo.create_member("Cara", "Martin", "cara.martin@example.com", "+155512340030", alto_cls.ClassificationId), ] for m in members: print(f" Member {m.MemberId}: {m.FirstName} {m.LastName} ({m.ClassificationId})") print("\n") print("=== 📦 Seeding Service Types & Availability ===") # ----------------------------------------------------------------- # 1️⃣ Service Types (keep the IDs for later use) # ----------------------------------------------------------------- st_9am = repo.create_service_type("9AM") st_11am = repo.create_service_type("11AM") st_6pm = repo.create_service_type("6PM") service_type_ids: Dict[str, int] = { "9AM": st_9am.ServiceTypeId, "11AM": st_11am.ServiceTypeId, "6PM": st_6pm.ServiceTypeId, } print( f"Created service types → " f"{st_9am.ServiceTypeId}=9AM, " f"{st_11am.ServiceTypeId}=11AM, " f"{st_6pm.ServiceTypeId}=6PM" ) # ----------------------------------------------------------------- # 2️⃣ Build a baseline availability map (member_id → set of ServiceTypeIds) # ----------------------------------------------------------------- def base_availability() -> Set[int]: """Return a set of ServiceTypeIds the member can take.""" roll = random.random() if roll < 0.30: # ~30 % get *all* three slots return set(service_type_ids.values()) elif roll < 0.70: # ~40 % get exactly two slots return set(random.sample(list(service_type_ids.values()), 2)) else: # ~30 % get a single slot return {random.choice(list(service_type_ids.values()))} # Populate the map for every member you created earlier availability_map: Dict[int, Set[int]] = {} for m in members: # `members` is the list you seeded above availability_map[m.MemberId] = base_availability() # ----------------------------------------------------------------- # 3️⃣ Hand‑crafted overrides (edge‑cases you want to guarantee) # ----------------------------------------------------------------- # Tenor block (IDs 1‑5 & 21‑25) → only 9AM & 11AM tenor_ids = [1, 2, 3, 4, 5, 21, 22, 23, 24, 25] for mid in tenor_ids: availability_map[mid] = { service_type_ids["9AM"], service_type_ids["11AM"], } # Baritone block (IDs 16‑20) → only 6PM baritone_ids = [16, 17, 18, 19, 20] for mid in baritone_ids: availability_map[mid] = {service_type_ids["6PM"]} # Ensure at least one member can do each slot (explicit adds) availability_map[1].add(service_type_ids["9AM"]) # John – Tenor → 9AM availability_map[6].add(service_type_ids["11AM"]) # Alice – Alto → 11AM availability_map[11].add(service_type_ids["6PM"]) # Bob – Soprano → 6PM # ----------------------------------------------------------------- # 4️⃣ Bulk‑insert into ServiceAvailability # ----------------------------------------------------------------- rows: List[Tuple[int, int]] = [] for member_id, type_set in availability_map.items(): for st_id in type_set: rows.append((member_id, st_id)) for row in rows: repo.db.execute( """ INSERT INTO ServiceAvailability (MemberId, ServiceTypeId) VALUES (?, ?) """, row, ) print( f"Inserted {len(rows)} ServiceAvailability rows " f"(≈ {len(members)} members × avg. {len(rows)//len(members)} slots each)." ) # ----- service (the day we are scheduling) --------------------------- service_dates = next_n_sundays(3) services = [] for service_date in service_dates: service = repo.create_service(st_6pm.ServiceTypeId, service_date) print(f"Created Service → ServiceId={service.ServiceId}, Date={service.ServiceDate}") services.append(service) print("\n") # -------------------------------------------------------------------- # 1️⃣ Get the first Tenor and ACCEPT it # -------------------------------------------------------------------- print("=== 🎯 FIRST SCHEDULE (should be John) ===") scheduled_member = repo.schedule_next_member( classification_id=soprano_cls.ClassificationId, service_id=services[0].ServiceId, only_active=True, ) print(scheduled_member) scheduled_member = repo.schedule_next_member( classification_id=tenor_cls.ClassificationId, service_id=services[0].ServiceId, only_active=True, ) print(scheduled_member) scheduled_member = repo.schedule_next_member( classification_id=tenor_cls.ClassificationId, service_id=services[0].ServiceId, only_active=True, ) print(scheduled_member) scheduled_member = repo.schedule_next_member( classification_id=tenor_cls.ClassificationId, service_id=services[0].ServiceId, only_active=True, ) print(scheduled_member) scheduled_member = repo.schedule_next_member( classification_id=tenor_cls.ClassificationId, service_id=services[0].ServiceId, only_active=True, ) print(scheduled_member) scheduled_member = repo.schedule_next_member( classification_id=tenor_cls.ClassificationId, service_id=services[2].ServiceId, only_active=True, ) print(scheduled_member) # ---------------------------------------------------------------------- # 2️⃣ Demo that exercises the full repository API # ---------------------------------------------------------------------- def demo(repo) -> None: return # ---------------------------------------------------------------------- # 5️⃣ Entrypoint # ---------------------------------------------------------------------- if __name__ == "__main__": # -------------------------------------------------------------- # Path to the SQLite file (feel free to change) # -------------------------------------------------------------- DB_PATH = Path(__file__).parent / "database_demo.db" # -------------------------------------------------------------- # Initialise DB if necessary # -------------------------------------------------------------- init_db(DB_PATH) exit() # -------------------------------------------------------------- # Build the connection / repository objects # -------------------------------------------------------------- from backend.database.connection import DatabaseConnection from backend.database.repository import Repository db = DatabaseConnection(str(DB_PATH)) repo = Repository(db) try: demo(repo) finally: # Always close the connection – SQLite locks the file while open db.close() print("\n✅ Demo finished – connection closed.")