feat: add UserService with CRUD operations

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

app/services/user_service.py

@@ -0,0 +1,128 @@
+"""Service layer for user profile management.
+
+All user database operations go through this service.
+Routes should never query the users table directly.
+"""
+
+from typing import Optional
+
+import structlog
+from sqlmodel import Session, select
+
+from app.models.user import User
+
+logger = structlog.get_logger(__name__)
+
+
+class UserService:
+    """Handles CRUD operations for User records.
+
+    Args:
+        session: An active SQLModel Session.
+    """
+
+    def __init__(self, session: Session) -> None:
+        self._session = session
+
+    def create_user(
+        self,
+        username: str,
+        password_hash: str,
+        display_name: str,
+        height: Optional[str] = None,
+        weight: Optional[str] = None,
+        goals: Optional[str] = None,
+        is_admin: bool = False,
+    ) -> User:
+        """Create a new user profile.
+
+        Args:
+            username: Unique login identifier.
+            password_hash: Pre-hashed password string.
+            display_name: Human-readable name.
+            height: User height as string.
+            weight: User weight as string.
+            goals: Free-text goals.
+            is_admin: Whether user has admin privileges.
+
+        Returns:
+            The newly created User record.
+        """
+        user = User(
+            username=username,
+            password_hash=password_hash,
+            display_name=display_name,
+            height=height,
+            weight=weight,
+            goals=goals,
+            is_admin=is_admin,
+        )
+        self._session.add(user)
+        self._session.commit()
+        self._session.refresh(user)
+        logger.info("user_created", username=username, is_admin=is_admin)
+        return user
+
+    def get_user_by_id(self, user_id: int) -> Optional[User]:
+        """Retrieve a user by primary key.
+
+        Args:
+            user_id: The user's ID.
+
+        Returns:
+            The User record, or None if not found.
+        """
+        return self._session.get(User, user_id)
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        """Retrieve a user by username.
+
+        Args:
+            username: The username to look up.
+
+        Returns:
+            The User record, or None if not found.
+        """
+        statement = select(User).where(User.username == username)
+        return self._session.exec(statement).first()
+
+    def list_users(self, exclude_admin: bool = False) -> list[User]:
+        """List all user profiles.
+
+        Args:
+            exclude_admin: If True, omit admin users from the result.
+
+        Returns:
+            List of User records.
+        """
+        statement = select(User)
+        if exclude_admin:
+            statement = statement.where(User.is_admin == False)  # noqa: E712
+        return list(self._session.exec(statement).all())
+
+    def update_user(self, user_id: int, **kwargs) -> User:
+        """Update fields on an existing user.
+
+        Args:
+            user_id: The user's ID.
+            **kwargs: Field names and new values to update.
+
+        Returns:
+            The updated User record.
+
+        Raises:
+            ValueError: If the user is not found.
+        """
+        user = self.get_user_by_id(user_id)
+        if user is None:
+            raise ValueError(f"User with id {user_id} not found")
+
+        for key, value in kwargs.items():
+            if hasattr(user, key):
+                setattr(user, key, value)
+
+        self._session.add(user)
+        self._session.commit()
+        self._session.refresh(user)
+        logger.info("user_updated", user_id=user_id, fields=list(kwargs.keys()))
+        return user

tests/test_user_service.py

@@ -0,0 +1,89 @@
+"""Tests for the UserService class."""
+
+from sqlmodel import SQLModel, Session, create_engine
+
+from app.models.user import User
+from app.services.user_service import UserService
+
+
+class TestUserService:
+    """Tests for user CRUD operations."""
+
+    def _setup(self):
+        """Create an in-memory DB and service instance."""
+        engine = create_engine("sqlite:///:memory:")
+        SQLModel.metadata.create_all(engine)
+        session = Session(engine)
+        service = UserService(session)
+        return session, service
+
+    def test_create_user(self) -> None:
+        """create_user should insert a new user and return it."""
+        session, service = self._setup()
+        user = service.create_user(
+            username="phil",
+            password_hash="hashed",
+            display_name="Phillip",
+            height="6'0\"",
+            weight="260 lbs",
+            goals="Muscle build",
+            is_admin=False,
+        )
+        assert user.id is not None
+        assert user.username == "phil"
+        session.close()
+
+    def test_get_user_by_id(self) -> None:
+        """get_user_by_id should return the correct user."""
+        session, service = self._setup()
+        created = service.create_user(
+            username="test", password_hash="h", display_name="Test"
+        )
+        found = service.get_user_by_id(created.id)
+        assert found is not None
+        assert found.username == "test"
+        session.close()
+
+    def test_get_user_by_username(self) -> None:
+        """get_user_by_username should return the correct user."""
+        session, service = self._setup()
+        service.create_user(username="admin", password_hash="h", display_name="Admin")
+        found = service.get_user_by_username("admin")
+        assert found is not None
+        assert found.display_name == "Admin"
+        session.close()
+
+    def test_get_user_by_username_not_found(self) -> None:
+        """get_user_by_username should return None for missing users."""
+        session, service = self._setup()
+        found = service.get_user_by_username("nonexistent")
+        assert found is None
+        session.close()
+
+    def test_list_users(self) -> None:
+        """list_users should return all users."""
+        session, service = self._setup()
+        service.create_user(username="a", password_hash="h", display_name="A")
+        service.create_user(username="b", password_hash="h", display_name="B")
+        users = service.list_users()
+        assert len(users) == 2
+        session.close()
+
+    def test_list_non_admin_users(self) -> None:
+        """list_users with exclude_admin=True should skip admin users."""
+        session, service = self._setup()
+        service.create_user(username="admin", password_hash="h", display_name="Admin", is_admin=True)
+        service.create_user(username="user", password_hash="h", display_name="User", is_admin=False)
+        users = service.list_users(exclude_admin=True)
+        assert len(users) == 1
+        assert users[0].username == "user"
+        session.close()
+
+    def test_update_user(self) -> None:
+        """update_user should modify the specified fields."""
+        session, service = self._setup()
+        user = service.create_user(username="u", password_hash="h", display_name="Old")
+        updated = service.update_user(user.id, display_name="New", weight="250 lbs")
+        assert updated.display_name == "New"
+        assert updated.weight == "250 lbs"
+        session.close()