feat: add SeedService for YAML-based database seeding

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app/services/seed_service.py

@@ -0,0 +1,233 @@
+"""Service for seeding the database from YAML config files.
+
+Reads config/exercises.yaml and config/user_programs.yaml to populate
+the exercise library, warmups, workout days, and user programs.
+"""
+
+import os
+from pathlib import Path
+from typing import Optional
+
+import bcrypt
+import structlog
+import yaml
+from sqlmodel import Session, select
+
+from app.models.exercise import Exercise
+from app.models.user import User
+from app.models.user_exercise_program import UserExerciseProgram
+from app.models.warmup import Warmup
+from app.models.workout_day import WorkoutDay
+
+logger = structlog.get_logger(__name__)
+
+# Default workout day definitions
+WORKOUT_DAYS = [
+    {"name": "Push", "day_number": 1, "description": "Chest, shoulders, triceps"},
+    {"name": "Pull", "day_number": 2, "description": "Back, biceps, traps"},
+    {"name": "Lower", "day_number": 3, "description": "Quads, hamstrings, glutes, calves"},
+    {"name": "Full Body", "day_number": 4, "description": "Compound full-body movements"},
+]
+
+
+class SeedService:
+    """Seeds the database from YAML configuration files.
+
+    Args:
+        session: An active SQLModel Session.
+        config_dir: Path to the config directory containing YAML files.
+    """
+
+    def __init__(self, session: Session, config_dir: Optional[Path] = None) -> None:
+        self._session = session
+        self._config_dir = config_dir or Path(__file__).resolve().parent.parent.parent / "config"
+
+    def _load_yaml(self, filename: str) -> dict:
+        """Load and parse a YAML file from the config directory.
+
+        Args:
+            filename: Name of the YAML file to load.
+
+        Returns:
+            Parsed YAML content as a dictionary.
+        """
+        filepath = self._config_dir / filename
+        with open(filepath, "r") as f:
+            return yaml.safe_load(f)
+
+    def seed_workout_days(self) -> None:
+        """Seed the 4 workout day records if they don't already exist."""
+        existing = self._session.exec(select(WorkoutDay)).first()
+        if existing:
+            logger.info("seed_skipped", table="workout_days", reason="already seeded")
+            return
+
+        for day_data in WORKOUT_DAYS:
+            day = WorkoutDay(**day_data)
+            self._session.add(day)
+        self._session.commit()
+        logger.info("seed_complete", table="workout_days", count=len(WORKOUT_DAYS))
+
+    def seed_exercises(self) -> None:
+        """Seed exercises from config/exercises.yaml if table is empty."""
+        existing = self._session.exec(select(Exercise)).first()
+        if existing:
+            logger.info("seed_skipped", table="exercises", reason="already seeded")
+            return
+
+        data = self._load_yaml("exercises.yaml")
+        exercises = data.get("exercises", [])
+        for ex in exercises:
+            exercise = Exercise(
+                name=ex["name"],
+                muscle_group=ex["muscle_group"],
+                workout_day=ex["workout_day"],
+                sets=ex.get("sets", 3),
+                tempo=ex.get("tempo", ""),
+                form_cues=ex.get("form_cues", "").strip(),
+            )
+            self._session.add(exercise)
+        self._session.commit()
+        logger.info("seed_complete", table="exercises", count=len(exercises))
+
+    def seed_warmups(self) -> None:
+        """Seed warmups from config/exercises.yaml if table is empty."""
+        existing = self._session.exec(select(Warmup)).first()
+        if existing:
+            logger.info("seed_skipped", table="warmups", reason="already seeded")
+            return
+
+        data = self._load_yaml("exercises.yaml")
+        warmups = data.get("warmup", [])
+        for i, wu in enumerate(warmups, start=1):
+            warmup = Warmup(
+                name=wu["name"],
+                type=wu.get("type", ""),
+                reps=wu.get("reps", ""),
+                form_cues=wu.get("form_cues", "").strip(),
+                sort_order=i,
+            )
+            self._session.add(warmup)
+        self._session.commit()
+        logger.info("seed_complete", table="warmups", count=len(warmups))
+
+    def seed_admin(self) -> None:
+        """Create admin user from environment variables if not exists.
+
+        Reads ADMIN_USERNAME and ADMIN_PASSWORD from env,
+        hashes the password with bcrypt, and creates the admin user.
+        """
+        admin_username = os.environ.get("ADMIN_USERNAME", "admin")
+        admin_password = os.environ.get("ADMIN_PASSWORD", "")
+
+        existing = self._session.exec(
+            select(User).where(User.username == admin_username)
+        ).first()
+        if existing:
+            logger.info("seed_skipped", table="users", reason="admin already exists")
+            return
+
+        if not admin_password:
+            logger.warning("seed_skipped", table="users", reason="ADMIN_PASSWORD not set")
+            return
+
+        # Hash password with bcrypt
+        password_hash = bcrypt.hashpw(
+            admin_password.encode("utf-8"),
+            bcrypt.gensalt(),
+        ).decode("utf-8")
+
+        admin = User(
+            username=admin_username,
+            password_hash=password_hash,
+            display_name="Admin",
+            is_admin=True,
+        )
+        self._session.add(admin)
+        self._session.commit()
+        logger.info("seed_complete", table="users", user="admin")
+
+    def seed_user_programs(self) -> None:
+        """Seed user profiles and their exercise programs from user_programs.yaml.
+
+        Creates non-admin user profiles and links them to exercises
+        with week 1/4 rep and weight targets.
+        """
+        data = self._load_yaml("user_programs.yaml")
+        programs = data.get("programs", [])
+
+        for program in programs:
+            username = program["user"].lower().replace(" ", "_")
+            display_name = program["user"]
+            profile = program.get("profile", {})
+
+            # Check if user already exists
+            existing_user = self._session.exec(
+                select(User).where(User.username == username)
+            ).first()
+
+            if existing_user:
+                user = existing_user
+                logger.info("seed_skipped", table="users", user=username, reason="already exists")
+            else:
+                user = User(
+                    username=username,
+                    password_hash="",  # Non-admin users don't log in initially
+                    display_name=display_name,
+                    height=profile.get("height", ""),
+                    weight=profile.get("weight", ""),
+                    goals=profile.get("goals", ""),
+                    is_admin=False,
+                )
+                self._session.add(user)
+                self._session.commit()
+                self._session.refresh(user)
+                logger.info("seed_complete", table="users", user=username)
+
+            # Link exercises to user
+            for ex_data in program.get("exercises", []):
+                exercise = self._session.exec(
+                    select(Exercise).where(Exercise.name == ex_data["name"])
+                ).first()
+
+                if exercise is None:
+                    logger.warning("seed_exercise_not_found", name=ex_data["name"])
+                    continue
+
+                # Check if program already exists
+                existing_program = self._session.exec(
+                    select(UserExerciseProgram).where(
+                        UserExerciseProgram.user_id == user.id,
+                        UserExerciseProgram.exercise_id == exercise.id,
+                    )
+                ).first()
+
+                if existing_program:
+                    continue
+
+                uep = UserExerciseProgram(
+                    user_id=user.id,
+                    exercise_id=exercise.id,
+                    wk1_reps=str(ex_data.get("wk1_reps", "")),
+                    wk4_reps=str(ex_data.get("wk4_reps", "")),
+                    wk1_weight=str(ex_data.get("wk1_weight", "")),
+                    wk4_weight=str(ex_data.get("wk4_weight", "")),
+                )
+                self._session.add(uep)
+
+            self._session.commit()
+            logger.info("seed_complete", table="user_exercise_programs", user=username)
+
+    def seed_all(self) -> None:
+        """Run all seed operations in the correct order.
+
+        Order matters: workout_days and exercises must exist before
+        user_programs can reference them.
+        """
+        logger.info("seed_all_started")
+        self.seed_workout_days()
+        self.seed_exercises()
+        self.seed_warmups()
+        self.seed_admin()
+        self.seed_user_programs()
+        logger.info("seed_all_complete")

tests/test_seed_service.py

@@ -0,0 +1,94 @@
+"""Tests for the SeedService class."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+import bcrypt
+from sqlmodel import SQLModel, Session, create_engine, select
+
+from app.models.user import User
+from app.models.exercise import Exercise
+from app.models.warmup import Warmup
+from app.models.workout_day import WorkoutDay
+from app.models.user_exercise_program import UserExerciseProgram
+from app.services.seed_service import SeedService
+
+
+class TestSeedService:
+    """Tests for YAML-based database seeding."""
+
+    def _setup(self):
+        """Create an in-memory DB and seed service."""
+        engine = create_engine("sqlite:///:memory:")
+        SQLModel.metadata.create_all(engine)
+        session = Session(engine)
+        # Use the real config files from the project
+        config_dir = Path(__file__).resolve().parent.parent / "config"
+        service = SeedService(session, config_dir=config_dir)
+        return session, service
+
+    def test_seed_workout_days(self) -> None:
+        """seed_workout_days should create 4 workout day records."""
+        session, service = self._setup()
+        service.seed_workout_days()
+        days = session.exec(select(WorkoutDay)).all()
+        assert len(days) == 4
+        names = {d.name for d in days}
+        assert names == {"Push", "Pull", "Lower", "Full Body"}
+        session.close()
+
+    def test_seed_exercises_from_yaml(self) -> None:
+        """seed_exercises should load all exercises from exercises.yaml."""
+        session, service = self._setup()
+        service.seed_exercises()
+        exercises = session.exec(select(Exercise)).all()
+        assert len(exercises) == 20  # 5 Push + 5 Pull + 5 Lower + 5 Full Body
+        session.close()
+
+    def test_seed_warmups_from_yaml(self) -> None:
+        """seed_warmups should load all warmups from exercises.yaml."""
+        session, service = self._setup()
+        service.seed_warmups()
+        warmups = session.exec(select(Warmup)).all()
+        assert len(warmups) == 6
+        session.close()
+
+    def test_seed_admin_user(self) -> None:
+        """seed_admin should create admin user with hashed password."""
+        session, service = self._setup()
+        with patch.dict("os.environ", {
+            "ADMIN_USERNAME": "admin",
+            "ADMIN_PASSWORD": "testpass",
+        }):
+            service.seed_admin()
+        admin = session.exec(select(User).where(User.is_admin == True)).first()  # noqa: E712
+        assert admin is not None
+        assert admin.username == "admin"
+        assert bcrypt.checkpw(b"testpass", admin.password_hash.encode())
+        session.close()
+
+    def test_seed_user_programs(self) -> None:
+        """seed_user_programs should create user profiles and link exercises."""
+        session, service = self._setup()
+        service.seed_exercises()
+        service.seed_user_programs()
+        programs = session.exec(select(UserExerciseProgram)).all()
+        assert len(programs) > 0
+        users = session.exec(select(User).where(User.is_admin == False)).all()  # noqa: E712
+        assert len(users) == 2  # Phillip and Daughter
+        session.close()
+
+    def test_seed_is_idempotent(self) -> None:
+        """Running seed_all twice should not create duplicate records."""
+        session, service = self._setup()
+        with patch.dict("os.environ", {
+            "ADMIN_USERNAME": "admin",
+            "ADMIN_PASSWORD": "testpass",
+        }):
+            service.seed_all()
+            service.seed_all()
+        users = session.exec(select(User)).all()
+        exercises = session.exec(select(Exercise)).all()
+        # Should not be doubled
+        assert len(exercises) == 20
+        session.close()