feat: add Phase 4 Logging & Tracking — inline set logging, history views
Add workout logging so users can track sets, reps, weight, and a
"felt easy?" toggle inline from the workout day view via HTMX.
Sessions auto-create on first log. History page shows past sessions
with detailed per-exercise breakdowns.
New services: WorkoutSessionService, LogService
New routes: POST /log, /log/{id}/edit, /log/{id}/delete, GET /history, /history/{id}
New templates: log_form, log_entry, session_card, log_history, session_detail
Modified: exercise_card (inline logging), nav (History link), workouts route (session context)
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
195
app/services/log_service.py
Normal file
195
app/services/log_service.py
Normal file
@@ -0,0 +1,195 @@
|
||||
"""Service layer for workout log (set-level) data access.
|
||||
|
||||
Handles CRUD for individual set logs within workout sessions.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.models.workout_log import WorkoutLog
|
||||
from app.models.workout_session import WorkoutSession
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class LogService:
|
||||
"""Handles CRUD operations for WorkoutLog records.
|
||||
|
||||
Args:
|
||||
session: An active SQLModel Session.
|
||||
"""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self._session = session
|
||||
|
||||
def create_log(
|
||||
self,
|
||||
session_id: int,
|
||||
exercise_id: int,
|
||||
set_number: int,
|
||||
reps_completed: int,
|
||||
weight_used: str,
|
||||
felt_easy: bool,
|
||||
notes: Optional[str] = None,
|
||||
) -> WorkoutLog:
|
||||
"""Create a new set log entry.
|
||||
|
||||
Args:
|
||||
session_id: FK to workout_sessions.
|
||||
exercise_id: FK to exercises.
|
||||
set_number: Which set (1, 2, 3...).
|
||||
reps_completed: Actual reps performed.
|
||||
weight_used: Weight as string (e.g., "30 lbs").
|
||||
felt_easy: Whether the set felt easy.
|
||||
notes: Optional notes for this set.
|
||||
|
||||
Returns:
|
||||
The newly created WorkoutLog record.
|
||||
"""
|
||||
log = WorkoutLog(
|
||||
session_id=session_id,
|
||||
exercise_id=exercise_id,
|
||||
set_number=set_number,
|
||||
reps_completed=reps_completed,
|
||||
weight_used=weight_used,
|
||||
felt_easy=felt_easy,
|
||||
notes=notes,
|
||||
)
|
||||
self._session.add(log)
|
||||
self._session.commit()
|
||||
self._session.refresh(log)
|
||||
logger.info(
|
||||
"log_created",
|
||||
session_id=session_id,
|
||||
exercise_id=exercise_id,
|
||||
set=set_number,
|
||||
)
|
||||
return log
|
||||
|
||||
def list_logs_for_session(self, session_id: int) -> list[WorkoutLog]:
|
||||
"""List all log entries for a workout session.
|
||||
|
||||
Args:
|
||||
session_id: The workout session ID.
|
||||
|
||||
Returns:
|
||||
List of WorkoutLog records ordered by exercise and set number.
|
||||
"""
|
||||
statement = (
|
||||
select(WorkoutLog)
|
||||
.where(WorkoutLog.session_id == session_id)
|
||||
.order_by(WorkoutLog.exercise_id, WorkoutLog.set_number)
|
||||
)
|
||||
return list(self._session.exec(statement).all())
|
||||
|
||||
def list_logs_for_exercise(
|
||||
self,
|
||||
session_id: int,
|
||||
exercise_id: int,
|
||||
) -> list[WorkoutLog]:
|
||||
"""List log entries for a specific exercise within a session.
|
||||
|
||||
Args:
|
||||
session_id: The workout session ID.
|
||||
exercise_id: The exercise ID.
|
||||
|
||||
Returns:
|
||||
List of WorkoutLog records for this exercise, ordered by set.
|
||||
"""
|
||||
statement = (
|
||||
select(WorkoutLog)
|
||||
.where(
|
||||
WorkoutLog.session_id == session_id,
|
||||
WorkoutLog.exercise_id == exercise_id,
|
||||
)
|
||||
.order_by(WorkoutLog.set_number)
|
||||
)
|
||||
return list(self._session.exec(statement).all())
|
||||
|
||||
def get_log_by_id(self, log_id: int) -> Optional[WorkoutLog]:
|
||||
"""Retrieve a log entry by primary key.
|
||||
|
||||
Args:
|
||||
log_id: The log entry ID.
|
||||
|
||||
Returns:
|
||||
The WorkoutLog record, or None if not found.
|
||||
"""
|
||||
return self._session.get(WorkoutLog, log_id)
|
||||
|
||||
def update_log(self, log_id: int, **kwargs) -> WorkoutLog:
|
||||
"""Update fields on an existing log entry.
|
||||
|
||||
Args:
|
||||
log_id: The log entry ID.
|
||||
**kwargs: Field names and new values.
|
||||
|
||||
Returns:
|
||||
The updated WorkoutLog record.
|
||||
|
||||
Raises:
|
||||
ValueError: If the log is not found.
|
||||
"""
|
||||
log = self.get_log_by_id(log_id)
|
||||
if log is None:
|
||||
raise ValueError(f"WorkoutLog with id {log_id} not found")
|
||||
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(log, key):
|
||||
setattr(log, key, value)
|
||||
|
||||
self._session.add(log)
|
||||
self._session.commit()
|
||||
self._session.refresh(log)
|
||||
logger.info("log_updated", log_id=log_id, fields=list(kwargs.keys()))
|
||||
return log
|
||||
|
||||
def delete_log(self, log_id: int) -> None:
|
||||
"""Delete a log entry.
|
||||
|
||||
Args:
|
||||
log_id: The log entry ID.
|
||||
|
||||
Raises:
|
||||
ValueError: If the log is not found.
|
||||
"""
|
||||
log = self.get_log_by_id(log_id)
|
||||
if log is None:
|
||||
raise ValueError(f"WorkoutLog with id {log_id} not found")
|
||||
|
||||
self._session.delete(log)
|
||||
self._session.commit()
|
||||
logger.info("log_deleted", log_id=log_id)
|
||||
|
||||
def get_latest_logs_for_exercise(
|
||||
self,
|
||||
user_id: int,
|
||||
exercise_id: int,
|
||||
limit: int = 10,
|
||||
) -> list[WorkoutLog]:
|
||||
"""Get the most recent log entries for an exercise across sessions.
|
||||
|
||||
Used by the progression engine (Phase 5) to determine
|
||||
what the user last did for this exercise.
|
||||
|
||||
Args:
|
||||
user_id: The user's ID.
|
||||
exercise_id: The exercise ID.
|
||||
limit: Maximum number of logs to return.
|
||||
|
||||
Returns:
|
||||
List of recent WorkoutLog records, newest first.
|
||||
"""
|
||||
statement = (
|
||||
select(WorkoutLog)
|
||||
.join(WorkoutSession, WorkoutLog.session_id == WorkoutSession.id)
|
||||
.where(
|
||||
WorkoutSession.user_id == user_id,
|
||||
WorkoutLog.exercise_id == exercise_id,
|
||||
)
|
||||
.order_by(WorkoutSession.date.desc(), WorkoutLog.set_number)
|
||||
.limit(limit)
|
||||
)
|
||||
return list(self._session.exec(statement).all())
|
||||
131
app/services/workout_session_service.py
Normal file
131
app/services/workout_session_service.py
Normal file
@@ -0,0 +1,131 @@
|
||||
"""Service layer for workout session management.
|
||||
|
||||
Handles creation, retrieval, and updates for workout sessions.
|
||||
A session represents a single workout on a specific date for a user.
|
||||
"""
|
||||
|
||||
from datetime import date
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.models.workout_session import WorkoutSession
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class WorkoutSessionService:
|
||||
"""Handles CRUD operations for WorkoutSession records.
|
||||
|
||||
Args:
|
||||
session: An active SQLModel Session.
|
||||
"""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self._session = session
|
||||
|
||||
def get_or_create_session(
|
||||
self,
|
||||
user_id: int,
|
||||
workout_day_id: int,
|
||||
session_date: date,
|
||||
) -> WorkoutSession:
|
||||
"""Get an existing session or create a new one.
|
||||
|
||||
If a session already exists for this user + day + date combo,
|
||||
return it. Otherwise, create a new one. This allows logging
|
||||
to start automatically without an explicit "start session" step.
|
||||
|
||||
Args:
|
||||
user_id: The user's ID.
|
||||
workout_day_id: The workout day's ID.
|
||||
session_date: The date of the workout.
|
||||
|
||||
Returns:
|
||||
The existing or newly created WorkoutSession.
|
||||
"""
|
||||
statement = select(WorkoutSession).where(
|
||||
WorkoutSession.user_id == user_id,
|
||||
WorkoutSession.workout_day_id == workout_day_id,
|
||||
WorkoutSession.date == session_date,
|
||||
)
|
||||
existing = self._session.exec(statement).first()
|
||||
|
||||
if existing:
|
||||
return existing
|
||||
|
||||
ws = WorkoutSession(
|
||||
user_id=user_id,
|
||||
workout_day_id=workout_day_id,
|
||||
date=session_date,
|
||||
)
|
||||
self._session.add(ws)
|
||||
self._session.commit()
|
||||
self._session.refresh(ws)
|
||||
logger.info(
|
||||
"workout_session_created",
|
||||
user_id=user_id,
|
||||
day_id=workout_day_id,
|
||||
date=str(session_date),
|
||||
)
|
||||
return ws
|
||||
|
||||
def list_sessions(
|
||||
self,
|
||||
user_id: int,
|
||||
limit: int = 50,
|
||||
) -> list[WorkoutSession]:
|
||||
"""List workout sessions for a user, most recent first.
|
||||
|
||||
Args:
|
||||
user_id: The user's ID.
|
||||
limit: Maximum number of sessions to return.
|
||||
|
||||
Returns:
|
||||
List of WorkoutSession records, ordered by date descending.
|
||||
"""
|
||||
statement = (
|
||||
select(WorkoutSession)
|
||||
.where(WorkoutSession.user_id == user_id)
|
||||
.order_by(WorkoutSession.date.desc())
|
||||
.limit(limit)
|
||||
)
|
||||
return list(self._session.exec(statement).all())
|
||||
|
||||
def get_session_by_id(self, session_id: int) -> Optional[WorkoutSession]:
|
||||
"""Retrieve a workout session by primary key.
|
||||
|
||||
Args:
|
||||
session_id: The session ID.
|
||||
|
||||
Returns:
|
||||
The WorkoutSession record, or None if not found.
|
||||
"""
|
||||
return self._session.get(WorkoutSession, session_id)
|
||||
|
||||
def update_session(self, session_id: int, **kwargs) -> WorkoutSession:
|
||||
"""Update fields on an existing workout session.
|
||||
|
||||
Args:
|
||||
session_id: The session ID.
|
||||
**kwargs: Field names and new values.
|
||||
|
||||
Returns:
|
||||
The updated WorkoutSession record.
|
||||
|
||||
Raises:
|
||||
ValueError: If the session is not found.
|
||||
"""
|
||||
ws = self.get_session_by_id(session_id)
|
||||
if ws is None:
|
||||
raise ValueError(f"WorkoutSession with id {session_id} not found")
|
||||
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(ws, key):
|
||||
setattr(ws, key, value)
|
||||
|
||||
self._session.add(ws)
|
||||
self._session.commit()
|
||||
self._session.refresh(ws)
|
||||
return ws
|
||||
Reference in New Issue
Block a user