sneakygeek/SneakySwole
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/phase2-data-layer'

Browse Source
This commit is contained in:
Phillip Tarrant
2026-02-24 10:15:10 -06:00
parent f9ff42c665 7c07bcee25
commit 1f47103480
24 changed files with 1753 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
alembic.ini Normal file
View File

@@ -0,0 +1,149 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts.
# this is typically a path given in POSIX (e.g. forward slashes)
# format, relative to the token %(here)s which refers to the location of this
# ini file
script_location = %(here)s/alembic
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# Or organize into date-based subdirectories (requires recursive_version_locations = true)
# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory. for multiple paths, the path separator
# is defined by "path_separator" below.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the tzdata library which can be installed by adding
# `alembic[tz]` to the pip requirements.
# string value is passed to ZoneInfo()
# leave blank for localtime
# timezone =
# max length of characters to apply to the "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to <script_location>/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "path_separator"
# below.
# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions
# path_separator; This indicates what character is used to split lists of file
# paths, including version_locations and prepend_sys_path within configparser
# files such as alembic.ini.
# The default rendered in new alembic.ini files is "os", which uses os.pathsep
# to provide os-dependent path splitting.
#
# Note that in order to support legacy alembic.ini files, this default does NOT
# take place if path_separator is not present in alembic.ini. If this
# option is omitted entirely, fallback logic is as follows:
#
# 1. Parsing of the version_locations option falls back to using the legacy
# "version_path_separator" key, which if absent then falls back to the legacy
# behavior of splitting on spaces and/or commas.
# 2. Parsing of the prepend_sys_path option falls back to the legacy
# behavior of splitting on spaces, commas, or colons.
#
# Valid values for path_separator are:
#
# path_separator = :
# path_separator = ;
# path_separator = space
# path_separator = newline
#
# Use os.pathsep. Default configuration used for new projects.
path_separator = os
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# database URL. This is consumed by the user-maintained env.py script only.
# other means of configuring database URLs may be customized within the env.py
# file.
sqlalchemy.url = sqlite:///data/sneakyswole.db
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
# hooks = ruff
# ruff.type = module
# ruff.module = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Alternatively, use the exec runner to execute a binary found on your PATH
# hooks = ruff
# ruff.type = exec
# ruff.executable = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Logging configuration. This is also consumed by the user-maintained
# env.py script only.
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

1
alembic/README Normal file
View File

@@ -0,0 +1 @@
Generic single-database configuration.

57
alembic/env.py Normal file
View File

@@ -0,0 +1,57 @@
"""Alembic environment configuration.
Imports all SQLModel models so autogenerate detects schema changes.
"""
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from sqlmodel import SQLModel
# Import all models so SQLModel.metadata knows about them
from app.models import ( # noqa: F401
User, Exercise, Warmup, WorkoutDay,
UserExerciseProgram, WorkoutSession, WorkoutLog, ProgressLog,
)
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = SQLModel.metadata
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode (SQL script generation)."""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode (direct DB connection)."""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

28
alembic/script.py.mako Normal file
View File

@@ -0,0 +1,28 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
def upgrade() -> None:
"""Upgrade schema."""
${upgrades if upgrades else "pass"}
def downgrade() -> None:
"""Downgrade schema."""
${downgrades if downgrades else "pass"}

155
alembic/versions/1855836abf6c_initial_schema_8_tables.py Normal file
View File

@@ -0,0 +1,155 @@
"""initial schema - 8 tables
Revision ID: 1855836abf6c
Revises:
Create Date: 2026-02-24 09:35:47.297957
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = '1855836abf6c'
down_revision: Union[str, Sequence[str], None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('exercises',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('muscle_group', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('workout_day', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('sets', sa.Integer(), nullable=False),
sa.Column('tempo', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('form_cues', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_exercises_name'), 'exercises', ['name'], unique=False)
op.create_index(op.f('ix_exercises_workout_day'), 'exercises', ['workout_day'], unique=False)
op.create_table('users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('username', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('password_hash', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('display_name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('height', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('weight', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('goals', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('is_admin', sa.Boolean(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
op.create_table('warmups',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('reps', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('form_cues', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('sort_order', sa.Integer(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_warmups_name'), 'warmups', ['name'], unique=False)
op.create_table('workout_days',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('day_number', sa.Integer(), nullable=False),
sa.Column('description', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('day_number')
)
op.create_index(op.f('ix_workout_days_name'), 'workout_days', ['name'], unique=True)
op.create_table('progress_log',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('exercise_id', sa.Integer(), nullable=False),
sa.Column('date', sa.Date(), nullable=False),
sa.Column('suggested_reps', sa.Integer(), nullable=True),
sa.Column('suggested_weight', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('actual_reps', sa.Integer(), nullable=True),
sa.Column('actual_weight', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('progression_applied', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['exercise_id'], ['exercises.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_progress_log_user_id'), 'progress_log', ['user_id'], unique=False)
op.create_table('user_exercise_programs',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('exercise_id', sa.Integer(), nullable=False),
sa.Column('wk1_reps', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('wk4_reps', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('wk1_weight', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('wk4_weight', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['exercise_id'], ['exercises.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_user_exercise_programs_exercise_id'), 'user_exercise_programs', ['exercise_id'], unique=False)
op.create_index(op.f('ix_user_exercise_programs_user_id'), 'user_exercise_programs', ['user_id'], unique=False)
op.create_table('workout_sessions',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('workout_day_id', sa.Integer(), nullable=False),
sa.Column('date', sa.Date(), nullable=False),
sa.Column('notes', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.ForeignKeyConstraint(['workout_day_id'], ['workout_days.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_workout_sessions_user_id'), 'workout_sessions', ['user_id'], unique=False)
op.create_table('workout_logs',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('session_id', sa.Integer(), nullable=False),
sa.Column('exercise_id', sa.Integer(), nullable=False),
sa.Column('set_number', sa.Integer(), nullable=False),
sa.Column('reps_completed', sa.Integer(), nullable=False),
sa.Column('weight_used', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('felt_easy', sa.Boolean(), nullable=False),
sa.Column('notes', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['exercise_id'], ['exercises.id'], ),
sa.ForeignKeyConstraint(['session_id'], ['workout_sessions.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_workout_logs_session_id'), 'workout_logs', ['session_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_workout_logs_session_id'), table_name='workout_logs')
op.drop_table('workout_logs')
op.drop_index(op.f('ix_workout_sessions_user_id'), table_name='workout_sessions')
op.drop_table('workout_sessions')
op.drop_index(op.f('ix_user_exercise_programs_user_id'), table_name='user_exercise_programs')
op.drop_index(op.f('ix_user_exercise_programs_exercise_id'), table_name='user_exercise_programs')
op.drop_table('user_exercise_programs')
op.drop_index(op.f('ix_progress_log_user_id'), table_name='progress_log')
op.drop_table('progress_log')
op.drop_index(op.f('ix_workout_days_name'), table_name='workout_days')
op.drop_table('workout_days')
op.drop_index(op.f('ix_warmups_name'), table_name='warmups')
op.drop_table('warmups')
op.drop_index(op.f('ix_users_username'), table_name='users')
op.drop_table('users')
op.drop_index(op.f('ix_exercises_workout_day'), table_name='exercises')
op.drop_index(op.f('ix_exercises_name'), table_name='exercises')
op.drop_table('exercises')
# ### end Alembic commands ###

60
app/database.py Normal file
View File

@@ -0,0 +1,60 @@
"""Database engine and session management.
Provides a SQLModel engine factory and a session dependency
for use with FastAPI's dependency injection.
"""
from pathlib import Path
from typing import Generator
import structlog
from sqlmodel import Session, create_engine
logger = structlog.get_logger(__name__)
def get_engine(database_url: str):
"""Create a SQLAlchemy engine for the given database URL.
For SQLite, ensures the parent directory exists and sets
appropriate connection arguments.
Args:
database_url: SQLAlchemy-compatible connection string.
Returns:
A configured SQLAlchemy Engine.
"""
connect_args = {}
if database_url.startswith("sqlite"):
# Ensure the data directory exists
db_path = database_url.replace("sqlite:///", "")
if db_path and db_path != ":memory:":
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
connect_args = {"check_same_thread": False}
engine = create_engine(
database_url,
connect_args=connect_args,
echo=False,
)
logger.info("database_engine_created", url=database_url)
return engine
def get_db_session(engine) -> Generator[Session, None, None]:
"""Yield a SQLModel session for dependency injection.
Usage in FastAPI routes:
@router.get("/example")
def example(session: Session = Depends(get_db_session)):
...
Args:
engine: The SQLAlchemy engine to bind the session to.
Yields:
A SQLModel Session that auto-closes after use.
"""
with Session(engine) as session:
yield session

21
app/main.py
View File

@@ -10,11 +10,14 @@ import structlog
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from sqlmodel import SQLModel, Session
from app.config import get_settings from app.config import get_settings
from app.database import get_engine, get_db_session
from app.logging_config import setup_logging from app.logging_config import setup_logging
from app.routes.health import router as health_router from app.routes.health import router as health_router
from app.routes.pages import router as pages_router from app.routes.pages import router as pages_router
from app.services.seed_service import SeedService
logger = structlog.get_logger(__name__) logger = structlog.get_logger(__name__)
@@ -50,6 +53,24 @@ def create_app() -> FastAPI:
app.include_router(health_router) app.include_router(health_router)
app.include_router(pages_router) app.include_router(pages_router)
# Database setup
engine = get_engine(settings.database_url)
SQLModel.metadata.create_all(engine)
app.state.engine = engine
# DB session dependency for routes
def _get_session():
yield from get_db_session(engine)
app.dependency_overrides[get_db_session] = _get_session
# Seed database on startup
@app.on_event("startup")
async def on_startup():
with Session(engine) as session:
seed_service = SeedService(session)
seed_service.seed_all()
logger.info("app_started", environment=settings.app_env) logger.info("app_started", environment=settings.app_env)
return app return app

24
app/models/__init__.py
View File

@@ -0,0 +1,24 @@
"""SQLModel model definitions for SneakySwole.
All 8 tables are defined here and re-exported for convenient imports.
"""
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.models.workout_session import WorkoutSession
from app.models.workout_log import WorkoutLog
from app.models.progress_log import ProgressLog
__all__ = [
"User",
"Exercise",
"Warmup",
"WorkoutDay",
"UserExerciseProgram",
"WorkoutSession",
"WorkoutLog",
"ProgressLog",
]

35
app/models/exercise.py Normal file
View File

@@ -0,0 +1,35 @@
"""Exercise model for the exercise library catalog.
Each exercise belongs to a workout day and includes form cues.
"""
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
class Exercise(SQLModel, table=True):
"""An exercise in the workout library.
Attributes:
id: Primary key, auto-incremented.
name: Exercise name (e.g., "DB Chest Press (Floor)").
muscle_group: Target muscle group (e.g., "Chest", "Shoulders").
workout_day: Which day this exercise belongs to (Push/Pull/Lower/Full Body).
sets: Default number of sets.
tempo: Tempo notation (e.g., "3-1-2" = 3s eccentric, 1s pause, 2s concentric).
form_cues: Detailed form instructions.
created_at: Timestamp when the record was created.
"""
__tablename__ = "exercises"
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
muscle_group: str = Field(default="")
workout_day: str = Field(index=True)
sets: int = Field(default=3)
tempo: str = Field(default="")
form_cues: str = Field(default="")
created_at: datetime = Field(default_factory=datetime.utcnow)

39
app/models/progress_log.py Normal file
View File

@@ -0,0 +1,39 @@
"""ProgressLog model for tracking progression suggestions and outcomes.
Records what the progression engine suggested vs what the user actually did.
"""
import datetime as dt
from typing import Optional
from sqlmodel import Field, SQLModel
class ProgressLog(SQLModel, table=True):
"""A progression tracking entry for a specific exercise.
Attributes:
id: Primary key, auto-incremented.
user_id: FK to users table.
exercise_id: FK to exercises table.
date: The date this progression entry applies to.
suggested_reps: What the engine recommended.
suggested_weight: What the engine recommended.
actual_reps: What the user actually did.
actual_weight: What the user actually used.
progression_applied: Type of progression (e.g., "reps_increase", "weight_increase", "deload").
created_at: Timestamp when the record was created.
"""
__tablename__ = "progress_log"
id: Optional[int] = Field(default=None, primary_key=True)
user_id: int = Field(foreign_key="users.id", index=True)
exercise_id: int = Field(foreign_key="exercises.id")
date: dt.date = Field(default_factory=dt.date.today)
suggested_reps: Optional[int] = Field(default=None)
suggested_weight: Optional[str] = Field(default=None)
actual_reps: Optional[int] = Field(default=None)
actual_weight: Optional[str] = Field(default=None)
progression_applied: Optional[str] = Field(default=None)
created_at: dt.datetime = Field(default_factory=dt.datetime.utcnow)

39
app/models/user.py Normal file
View File

@@ -0,0 +1,39 @@
"""User model for profile management.
Stores admin and regular user profiles with physical stats and goals.
"""
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
class User(SQLModel, table=True):
"""A user profile in the system.
Attributes:
id: Primary key, auto-incremented.
username: Unique login identifier.
password_hash: bcrypt-hashed password (admin only initially).
display_name: Human-readable name shown in the UI.
height: User's height as a string (e.g., "6'0\"").
weight: User's weight as a string (e.g., "260 lbs").
goals: Free-text training goals.
is_admin: Whether this user has admin privileges.
created_at: Timestamp when the record was created.
updated_at: Timestamp of the last update.
"""
__tablename__ = "users"
id: Optional[int] = Field(default=None, primary_key=True)
username: str = Field(index=True, unique=True)
password_hash: str = Field(default="")
display_name: str = Field(default="")
height: Optional[str] = Field(default=None)
weight: Optional[str] = Field(default=None)
goals: Optional[str] = Field(default=None)
is_admin: bool = Field(default=False)
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)

37
app/models/user_exercise_program.py Normal file
View File

@@ -0,0 +1,37 @@
"""UserExerciseProgram model for per-user exercise programming.
Links a user to an exercise with week 1 and week 4 rep/weight targets.
"""
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
class UserExerciseProgram(SQLModel, table=True):
"""Per-user programming for a specific exercise.
Attributes:
id: Primary key, auto-incremented.
user_id: FK to users table.
exercise_id: FK to exercises table.
wk1_reps: Week 1 target reps (string to support "30 sec" style).
wk4_reps: Week 4 target reps.
wk1_weight: Week 1 target weight (e.g., "30 lbs", "BW").
wk4_weight: Week 4 target weight.
created_at: Timestamp when the record was created.
updated_at: Timestamp of the last update.
"""
__tablename__ = "user_exercise_programs"
id: Optional[int] = Field(default=None, primary_key=True)
user_id: int = Field(foreign_key="users.id", index=True)
exercise_id: int = Field(foreign_key="exercises.id", index=True)
wk1_reps: str = Field(default="")
wk4_reps: str = Field(default="")
wk1_weight: str = Field(default="")
wk4_weight: str = Field(default="")
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)

33
app/models/warmup.py Normal file
View File

@@ -0,0 +1,33 @@
"""Warmup model for the standardized warmup routine.
Warmups are displayed before every workout in a fixed order.
"""
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
class Warmup(SQLModel, table=True):
"""A warmup exercise in the standardized routine.
Attributes:
id: Primary key, auto-incremented.
name: Warmup name (e.g., "Cat / Cow").
type: Category (e.g., "Thoracic Mob", "Hip Mobility").
reps: Rep scheme as a string (e.g., "8 reps", "8 each side").
form_cues: Detailed form instructions.
sort_order: Display order in the warmup sequence.
created_at: Timestamp when the record was created.
"""
__tablename__ = "warmups"
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
type: str = Field(default="")
reps: str = Field(default="")
form_cues: str = Field(default="")
sort_order: int = Field(default=0)
created_at: datetime = Field(default_factory=datetime.utcnow)

26
app/models/workout_day.py Normal file
View File

@@ -0,0 +1,26 @@
"""WorkoutDay model for the 4-day training split.
Defines the named workout days and their order.
"""
from typing import Optional
from sqlmodel import Field, SQLModel
class WorkoutDay(SQLModel, table=True):
"""A named workout day in the training program.
Attributes:
id: Primary key, auto-incremented.
name: Day name (Push, Pull, Lower, Full Body).
day_number: Order in the weekly rotation (1-4).
description: Brief description of the day's focus.
"""
__tablename__ = "workout_days"
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True, unique=True)
day_number: int = Field(unique=True)
description: str = Field(default="")

37
app/models/workout_log.py Normal file
View File

@@ -0,0 +1,37 @@
"""WorkoutLog model for per-exercise set logging.
Each log entry records one set of one exercise within a session.
"""
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
class WorkoutLog(SQLModel, table=True):
"""A single set log for an exercise within a workout session.
Attributes:
id: Primary key, auto-incremented.
session_id: FK to workout_sessions table.
exercise_id: FK to exercises table.
set_number: Which set this is (1, 2, 3...).
reps_completed: Actual reps performed.
weight_used: Weight used as string (e.g., "30 lbs", "BW").
felt_easy: Whether the user felt the set was easy (progression signal).
notes: Optional notes about this specific set.
created_at: Timestamp when the record was created.
"""
__tablename__ = "workout_logs"
id: Optional[int] = Field(default=None, primary_key=True)
session_id: int = Field(foreign_key="workout_sessions.id", index=True)
exercise_id: int = Field(foreign_key="exercises.id")
set_number: int = Field(default=1)
reps_completed: int = Field(default=0)
weight_used: str = Field(default="")
felt_easy: bool = Field(default=False)
notes: Optional[str] = Field(default=None)
created_at: datetime = Field(default_factory=datetime.utcnow)

31
app/models/workout_session.py Normal file
View File

@@ -0,0 +1,31 @@
"""WorkoutSession model for tracking completed workout sessions.
Each session ties a user to a workout day on a specific date.
"""
import datetime as dt
from typing import Optional
from sqlmodel import Field, SQLModel
class WorkoutSession(SQLModel, table=True):
"""A completed workout session.
Attributes:
id: Primary key, auto-incremented.
user_id: FK to users table.
workout_day_id: FK to workout_days table.
date: The date the workout was performed.
notes: Optional free-text notes about the session.
created_at: Timestamp when the record was created.
"""
__tablename__ = "workout_sessions"
id: Optional[int] = Field(default=None, primary_key=True)
user_id: int = Field(foreign_key="users.id", index=True)
workout_day_id: int = Field(foreign_key="workout_days.id")
date: dt.date = Field(default_factory=dt.date.today)
notes: Optional[str] = Field(default=None)
created_at: dt.datetime = Field(default_factory=dt.datetime.utcnow)

88
app/services/exercise_service.py Normal file
View File

@@ -0,0 +1,88 @@
"""Service layer for exercise, warmup, and workout day data access.
All exercise-related database operations go through this service.
"""
from typing import Optional
import structlog
from sqlmodel import Session, select
from app.models.exercise import Exercise
from app.models.warmup import Warmup
from app.models.workout_day import WorkoutDay
logger = structlog.get_logger(__name__)
class ExerciseService:
"""Handles read operations for exercises, warmups, and workout days.
Args:
session: An active SQLModel Session.
"""
def __init__(self, session: Session) -> None:
self._session = session
def list_exercises(
self,
workout_day: Optional[str] = None,
muscle_group: Optional[str] = None,
) -> list[Exercise]:
"""List exercises with optional filtering.
Args:
workout_day: Filter by workout day name (e.g., "Push").
muscle_group: Filter by muscle group (e.g., "Chest").
Returns:
List of matching Exercise records.
"""
statement = select(Exercise)
if workout_day:
statement = statement.where(Exercise.workout_day == workout_day)
if muscle_group:
statement = statement.where(Exercise.muscle_group == muscle_group)
return list(self._session.exec(statement).all())
def get_exercise_by_id(self, exercise_id: int) -> Optional[Exercise]:
"""Retrieve an exercise by primary key.
Args:
exercise_id: The exercise ID.
Returns:
The Exercise record, or None if not found.
"""
return self._session.get(Exercise, exercise_id)
def get_exercise_by_name(self, name: str) -> Optional[Exercise]:
"""Retrieve an exercise by exact name match.
Args:
name: The exercise name to look up.
Returns:
The Exercise record, or None if not found.
"""
statement = select(Exercise).where(Exercise.name == name)
return self._session.exec(statement).first()
def list_warmups(self) -> list[Warmup]:
"""List all warmups in display order.
Returns:
List of Warmup records sorted by sort_order.
"""
statement = select(Warmup).order_by(Warmup.sort_order)
return list(self._session.exec(statement).all())
def list_workout_days(self) -> list[WorkoutDay]:
"""List all workout days in order.
Returns:
List of WorkoutDay records sorted by day_number.
"""
statement = select(WorkoutDay).order_by(WorkoutDay.day_number)
return list(self._session.exec(statement).all())

233
app/services/seed_service.py Normal file
View File

@@ -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")

128
app/services/user_service.py Normal file
View File

@@ -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

26
tests/test_database.py Normal file
View File

@@ -0,0 +1,26 @@
"""Tests for database engine and session management."""
from sqlmodel import Session
from app.database import get_engine, get_db_session
class TestDatabase:
"""Tests for database connection management."""
def test_get_engine_returns_engine(self) -> None:
"""get_engine should return a SQLAlchemy engine instance."""
engine = get_engine("sqlite:///data/test.db")
assert engine is not None
assert "sqlite" in str(engine.url)
def test_get_db_session_yields_session(self) -> None:
"""get_db_session should yield a usable SQLModel Session."""
engine = get_engine("sqlite:///:memory:")
session_gen = get_db_session(engine)
session = next(session_gen)
assert isinstance(session, Session)
try:
next(session_gen)
except StopIteration:
pass

95
tests/test_exercise_service.py Normal file
View File

@@ -0,0 +1,95 @@
"""Tests for the ExerciseService class."""
from sqlmodel import SQLModel, Session, create_engine
from app.models.exercise import Exercise
from app.models.warmup import Warmup
from app.models.workout_day import WorkoutDay
from app.services.exercise_service import ExerciseService
class TestExerciseService:
"""Tests for exercise and warmup data access."""
def _setup(self):
"""Create an in-memory DB with seed data and return the service."""
engine = create_engine("sqlite:///:memory:")
SQLModel.metadata.create_all(engine)
session = Session(engine)
service = ExerciseService(session)
# Seed workout days
for i, (name, desc) in enumerate([
("Push", "Chest, shoulders, triceps"),
("Pull", "Back, biceps, traps"),
("Lower", "Quads, hamstrings, glutes, calves"),
("Full Body", "Compound movements"),
], start=1):
session.add(WorkoutDay(name=name, day_number=i, description=desc))
# Seed a few exercises
session.add(Exercise(name="DB Chest Press", muscle_group="Chest", workout_day="Push", sets=3, tempo="3-1-2", form_cues="..."))
session.add(Exercise(name="DB Row", muscle_group="Back", workout_day="Pull", sets=3, tempo="3-1-2", form_cues="..."))
# Seed warmups
session.add(Warmup(name="Cat / Cow", type="Thoracic Mob", reps="8 reps", form_cues="...", sort_order=1))
session.add(Warmup(name="Fire Hydrant", type="Hip Mobility", reps="8 each", form_cues="...", sort_order=2))
session.commit()
return session, service
def test_list_exercises_all(self) -> None:
"""list_exercises should return all exercises."""
session, service = self._setup()
exercises = service.list_exercises()
assert len(exercises) == 2
session.close()
def test_list_exercises_by_workout_day(self) -> None:
"""list_exercises should filter by workout day."""
session, service = self._setup()
exercises = service.list_exercises(workout_day="Push")
assert len(exercises) == 1
assert exercises[0].name == "DB Chest Press"
session.close()
def test_list_exercises_by_muscle_group(self) -> None:
"""list_exercises should filter by muscle group."""
session, service = self._setup()
exercises = service.list_exercises(muscle_group="Back")
assert len(exercises) == 1
session.close()
def test_get_exercise_by_id(self) -> None:
"""get_exercise_by_id should return the correct exercise."""
session, service = self._setup()
exercises = service.list_exercises()
found = service.get_exercise_by_id(exercises[0].id)
assert found is not None
session.close()
def test_get_exercise_by_name(self) -> None:
"""get_exercise_by_name should return the correct exercise."""
session, service = self._setup()
found = service.get_exercise_by_name("DB Chest Press")
assert found is not None
assert found.muscle_group == "Chest"
session.close()
def test_list_warmups_ordered(self) -> None:
"""list_warmups should return warmups in sort_order."""
session, service = self._setup()
warmups = service.list_warmups()
assert len(warmups) == 2
assert warmups[0].name == "Cat / Cow"
assert warmups[1].name == "Fire Hydrant"
session.close()
def test_list_workout_days(self) -> None:
"""list_workout_days should return all 4 days in order."""
session, service = self._setup()
days = service.list_workout_days()
assert len(days) == 4
assert days[0].name == "Push"
assert days[3].name == "Full Body"
session.close()

228
tests/test_models.py Normal file
View File

@@ -0,0 +1,228 @@
"""Tests for SQLModel model definitions."""
from datetime import datetime
from sqlmodel import SQLModel, Session, create_engine
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.models.workout_session import WorkoutSession
from app.models.workout_log import WorkoutLog
from app.models.progress_log import ProgressLog
class TestModels:
"""Tests that all models can be instantiated and persisted."""
def _create_engine(self):
"""Create an in-memory SQLite engine with all tables."""
engine = create_engine("sqlite:///:memory:")
SQLModel.metadata.create_all(engine)
return engine
def test_user_model_roundtrip(self) -> None:
"""User model should persist and retrieve correctly."""
engine = self._create_engine()
user = User(
username="testuser",
password_hash="fakehash",
display_name="Test User",
height="6'0\"",
weight="200 lbs",
goals="Get strong",
is_admin=False,
)
with Session(engine) as session:
session.add(user)
session.commit()
session.refresh(user)
assert user.id is not None
assert user.username == "testuser"
assert user.is_admin is False
assert isinstance(user.created_at, datetime)
def test_exercise_model_roundtrip(self) -> None:
"""Exercise model should persist and retrieve correctly."""
engine = self._create_engine()
exercise = Exercise(
name="DB Chest Press (Floor)",
muscle_group="Chest",
workout_day="Push",
sets=3,
tempo="3-1-2",
form_cues="Lie on the floor...",
)
with Session(engine) as session:
session.add(exercise)
session.commit()
session.refresh(exercise)
assert exercise.id is not None
assert exercise.name == "DB Chest Press (Floor)"
def test_warmup_model_roundtrip(self) -> None:
"""Warmup model should persist and retrieve correctly."""
engine = self._create_engine()
warmup = Warmup(
name="Cat / Cow",
type="Thoracic Mob",
reps="8 reps",
form_cues="Start on hands and knees...",
sort_order=1,
)
with Session(engine) as session:
session.add(warmup)
session.commit()
session.refresh(warmup)
assert warmup.id is not None
def test_workout_day_model_roundtrip(self) -> None:
"""WorkoutDay model should persist and retrieve correctly."""
engine = self._create_engine()
day = WorkoutDay(
name="Push",
day_number=1,
description="Chest, shoulders, triceps",
)
with Session(engine) as session:
session.add(day)
session.commit()
session.refresh(day)
assert day.id is not None
assert day.day_number == 1
def test_user_exercise_program_model_roundtrip(self) -> None:
"""UserExerciseProgram model should persist with FK references."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
exercise = Exercise(
name="Test Ex", muscle_group="Test",
workout_day="Push", sets=3, tempo="3-1-2", form_cues="..."
)
session.add(user)
session.add(exercise)
session.commit()
session.refresh(user)
session.refresh(exercise)
program = UserExerciseProgram(
user_id=user.id,
exercise_id=exercise.id,
wk1_reps="8",
wk4_reps="12",
wk1_weight="30 lbs",
wk4_weight="40 lbs",
)
session.add(program)
session.commit()
session.refresh(program)
assert program.id is not None
assert program.user_id == user.id
def test_workout_session_model_roundtrip(self) -> None:
"""WorkoutSession model should persist correctly."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
session.add(user)
session.add(day)
session.commit()
session.refresh(user)
session.refresh(day)
ws = WorkoutSession(
user_id=user.id,
workout_day_id=day.id,
date=datetime.utcnow().date(),
)
session.add(ws)
session.commit()
session.refresh(ws)
assert ws.id is not None
def test_workout_log_model_roundtrip(self) -> None:
"""WorkoutLog model should persist correctly."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
exercise = Exercise(
name="Ex", muscle_group="Test",
workout_day="Push", sets=3, tempo="3-1-2", form_cues="..."
)
session.add_all([user, day, exercise])
session.commit()
session.refresh(user)
session.refresh(day)
session.refresh(exercise)
ws = WorkoutSession(
user_id=user.id,
workout_day_id=day.id,
date=datetime.utcnow().date(),
)
session.add(ws)
session.commit()
session.refresh(ws)
log = WorkoutLog(
session_id=ws.id,
exercise_id=exercise.id,
set_number=1,
reps_completed=8,
weight_used="30 lbs",
felt_easy=False,
)
session.add(log)
session.commit()
session.refresh(log)
assert log.id is not None
assert log.felt_easy is False
def test_progress_log_model_roundtrip(self) -> None:
"""ProgressLog model should persist correctly."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
exercise = Exercise(
name="Ex", muscle_group="Test",
workout_day="Push", sets=3, tempo="3-1-2", form_cues="..."
)
session.add_all([user, exercise])
session.commit()
session.refresh(user)
session.refresh(exercise)
pl = ProgressLog(
user_id=user.id,
exercise_id=exercise.id,
date=datetime.utcnow().date(),
suggested_reps=10,
suggested_weight="35 lbs",
actual_reps=10,
actual_weight="35 lbs",
progression_applied="reps_increase",
)
session.add(pl)
session.commit()
session.refresh(pl)
assert pl.id is not None
def test_all_models_importable_from_init(self) -> None:
"""All models should be importable from app.models."""
from app.models import (
User, Exercise, Warmup, WorkoutDay,
UserExerciseProgram, WorkoutSession, WorkoutLog, ProgressLog,
)
assert User is not None
assert Exercise is not None
assert Warmup is not None
assert WorkoutDay is not None
assert UserExerciseProgram is not None
assert WorkoutSession is not None
assert WorkoutLog is not None
assert ProgressLog is not None

94
tests/test_seed_service.py Normal file
View File

@@ -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()

89
tests/test_user_service.py Normal file
View File

@@ -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()