dd82f461be
- Create database tables: workout_program, program_session, person_program_assignment.
- Add Flask blueprint `routes/programs.py` with routes for creating, listing, viewing, and deleting programs.
- Implement program creation form (`templates/program_create.html`):
- Allows defining program name, description, and multiple sessions.
- Each session includes a name and dynamically added exercise selections.
- Uses `tail.select` for searchable exercise dropdowns.
- JavaScript handles dynamic addition/removal of sessions and exercises.
- Implement backend logic for program creation:
- Parses form data including multiple exercises per session.
- Automatically finds or creates non-person-specific tags based on selected exercises for each session.
- Saves program and session data, linking sessions to appropriate tags.
- Implement program list view (`templates/program_list.html`):
- Displays existing programs.
- Includes HTMX-enabled delete button for each program.
- Links program names to the view page using HTMX for dynamic loading.
- Implement program detail view (`templates/program_view.html`):
- Displays program name, description, and sessions.
- Parses session tag filters to retrieve and display associated exercises.
- Update changelog with details of the new feature.
271 lines
13 KiB
Python
271 lines
13 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, current_app
|
|
from extensions import db
|
|
# from flask_login import login_required, current_user # Add if authentication is needed
|
|
from jinja2_fragments import render_block # Import render_block
|
|
|
|
programs_bp = Blueprint('programs', __name__, url_prefix='/programs')
|
|
|
|
from flask import flash # Import flash for displaying messages
|
|
|
|
@programs_bp.route('/create', methods=['GET', 'POST'])
|
|
# @login_required # Uncomment if login is required
|
|
def create_program():
|
|
if request.method == 'POST':
|
|
program_name = request.form.get('program_name', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
sessions_data = []
|
|
i = 0
|
|
while True:
|
|
# Check for the presence of session order to determine if the session exists
|
|
session_order_key = f'session_order_{i}'
|
|
if session_order_key not in request.form:
|
|
break # No more sessions
|
|
|
|
session_order = request.form.get(session_order_key)
|
|
session_name = request.form.get(f'session_name_{i}', '').strip()
|
|
# Get list of selected exercise IDs for this session
|
|
exercise_ids_str = request.form.getlist(f'exercises_{i}')
|
|
|
|
# Basic validation for session data
|
|
if not exercise_ids_str or not session_order:
|
|
flash(f"Error processing session {i+1}: Missing exercises or order.", "error")
|
|
# TODO: Re-render form preserving entered data
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
try:
|
|
# Convert exercise IDs to integers and sort them for consistent filter generation
|
|
exercise_ids = sorted([int(eid) for eid in exercise_ids_str])
|
|
|
|
sessions_data.append({
|
|
'order': int(session_order),
|
|
'name': session_name if session_name else None, # Store None if empty
|
|
'exercise_ids': exercise_ids # Store the list of exercise IDs
|
|
})
|
|
except ValueError:
|
|
flash(f"Error processing session {i+1}: Invalid exercise ID or order.", "error")
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
i += 1
|
|
|
|
# --- Validation ---
|
|
if not program_name:
|
|
flash("Program Name is required.", "error")
|
|
# TODO: Re-render form preserving entered data
|
|
return redirect(url_for('programs.create_program'))
|
|
if not sessions_data:
|
|
flash("At least one session must be added.", "error")
|
|
# TODO: Re-render form preserving entered data
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
# --- Database Insertion ---
|
|
try:
|
|
# Insert Program
|
|
program_result = db.execute(
|
|
"INSERT INTO workout_program (name, description) VALUES (%s, %s) RETURNING program_id",
|
|
[program_name, description if description else None],
|
|
commit=True, one=True
|
|
)
|
|
if not program_result or 'program_id' not in program_result:
|
|
raise Exception("Failed to create workout program entry.")
|
|
|
|
new_program_id = program_result['program_id']
|
|
|
|
# Insert Sessions (and find/create tags)
|
|
for session in sessions_data:
|
|
# 1. Generate the canonical filter string from sorted exercise IDs
|
|
if not session['exercise_ids']:
|
|
flash(f"Session {session['order']} must have at least one exercise selected.", "error")
|
|
# Ideally, rollback program insert or handle differently
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
tag_filter = "?" + "&".join(f"exercise_id={eid}" for eid in session['exercise_ids'])
|
|
tag_name = session['name'] if session['name'] else f"Program Session {session['order']} Exercises" # Default tag name
|
|
|
|
# 2. Find existing tag with this exact filter (non-person specific)
|
|
existing_tag = db.execute(
|
|
"SELECT tag_id FROM tag WHERE filter = %s AND person_id IS NULL",
|
|
[tag_filter], one=True
|
|
)
|
|
|
|
session_tag_id = None
|
|
if existing_tag:
|
|
session_tag_id = existing_tag['tag_id']
|
|
# Optional: Update tag name if session name provided and different?
|
|
# db.execute("UPDATE tag SET name = %s WHERE tag_id = %s", [tag_name, session_tag_id], commit=True)
|
|
else:
|
|
# 3. Create new tag if not found
|
|
# Ensure tag name uniqueness if desired (e.g., append number if name exists)
|
|
# For simplicity, allow duplicate names for now, rely on filter for uniqueness
|
|
new_tag_result = db.execute(
|
|
"INSERT INTO tag (name, filter, person_id) VALUES (%s, %s, NULL) RETURNING tag_id",
|
|
[tag_name, tag_filter], commit=True, one=True
|
|
)
|
|
if not new_tag_result or 'tag_id' not in new_tag_result:
|
|
raise Exception(f"Failed to create tag for session {session['order']}.")
|
|
session_tag_id = new_tag_result['tag_id']
|
|
|
|
# 4. Insert program_session using the found/created tag_id
|
|
db.execute(
|
|
"""INSERT INTO program_session (program_id, session_order, session_name, tag_id)
|
|
VALUES (%s, %s, %s, %s)""",
|
|
[new_program_id, session['order'], session['name'], session_tag_id],
|
|
commit=True # Commit each session insert
|
|
)
|
|
|
|
flash(f"Workout Program '{program_name}' created successfully!", "success")
|
|
# TODO: Redirect to a program view page once it exists
|
|
# return redirect(url_for('programs.view_program', program_id=new_program_id))
|
|
return redirect(url_for('programs.list_programs')) # Redirect to a list page for now
|
|
|
|
except Exception as e:
|
|
# Log the error e
|
|
print(f"Error creating program: {e}") # Basic logging
|
|
flash(f"Database error creating program: {e}", "error")
|
|
# Rollback might be needed if using transactions across inserts
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
else: # GET Request
|
|
# Fetch all available exercises to populate multi-selects
|
|
exercises = db.execute("SELECT exercise_id, name FROM exercise ORDER BY name")
|
|
if exercises is None:
|
|
exercises = [] # Ensure exercises is an iterable
|
|
|
|
# Pass exercises to the template context
|
|
return render_template('program_create.html', exercises=exercises, render_block=render_block) # Pass exercises instead of tags
|
|
|
|
|
|
from flask_htmx import HTMX # Import HTMX
|
|
|
|
htmx = HTMX() # Initialize HTMX if not already done globally
|
|
|
|
# Placeholder for program list route (used in POST redirect)
|
|
@programs_bp.route('/', methods=['GET'])
|
|
# @login_required
|
|
def list_programs():
|
|
# Fetch and display list of programs
|
|
programs = db.execute("SELECT program_id, name, description FROM workout_program ORDER BY created_at DESC")
|
|
if programs is None:
|
|
programs = []
|
|
|
|
# Check if it's an HTMX request
|
|
if htmx:
|
|
# Render only the content block for HTMX requests
|
|
return render_block(current_app.jinja_env, 'program_list.html', 'content', programs=programs)
|
|
else:
|
|
# Render the full page for regular requests
|
|
return render_template('program_list.html', programs=programs)
|
|
|
|
|
|
@programs_bp.route('/<int:program_id>/delete', methods=['DELETE'])
|
|
# @login_required # Add authentication if needed
|
|
def delete_program(program_id):
|
|
"""Deletes a workout program and its associated sessions/assignments."""
|
|
try:
|
|
# The ON DELETE CASCADE constraint on program_session and person_program_assignment
|
|
# should handle deleting related rows automatically when the program is deleted.
|
|
result = db.execute(
|
|
"DELETE FROM workout_program WHERE program_id = %s RETURNING program_id",
|
|
[program_id],
|
|
commit=True, one=True
|
|
)
|
|
if result and result.get('program_id') == program_id:
|
|
# Return empty response for HTMX, maybe trigger list refresh
|
|
# flash(f"Program ID {program_id} deleted successfully.", "success") # Flash might not show on empty response
|
|
response = "" # Empty response indicates success to HTMX
|
|
headers = {"HX-Trigger": "programDeleted"} # Trigger event for potential list refresh
|
|
return response, 200, headers
|
|
else:
|
|
# Program not found or delete failed silently
|
|
flash(f"Could not find or delete program ID {program_id}.", "error")
|
|
# Returning an error status might be better for HTMX error handling
|
|
return "Error: Program not found or deletion failed", 404
|
|
|
|
except Exception as e:
|
|
# Log the error e
|
|
print(f"Error deleting program {program_id}: {e}")
|
|
flash(f"Database error deleting program: {e}", "error")
|
|
# Return an error status for HTMX
|
|
return "Server error during deletion", 500
|
|
|
|
|
|
# TODO: Add routes for viewing, editing, and assigning programs
|
|
from urllib.parse import parse_qs # Needed to parse tag filters
|
|
|
|
@programs_bp.route('/<int:program_id>', methods=['GET'])
|
|
# @login_required
|
|
def view_program(program_id):
|
|
"""Displays the details of a specific workout program."""
|
|
# Fetch program details
|
|
program = db.execute(
|
|
"SELECT program_id, name, description, created_at FROM workout_program WHERE program_id = %s",
|
|
[program_id], one=True
|
|
)
|
|
if not program:
|
|
flash(f"Workout Program with ID {program_id} not found.", "error")
|
|
return redirect(url_for('programs.list_programs'))
|
|
|
|
# Fetch sessions and their associated tags
|
|
sessions = db.execute(
|
|
"""
|
|
SELECT
|
|
ps.session_id, ps.session_order, ps.session_name,
|
|
t.tag_id, t.name as tag_name, t.filter as tag_filter
|
|
FROM program_session ps
|
|
JOIN tag t ON ps.tag_id = t.tag_id
|
|
WHERE ps.program_id = %s
|
|
ORDER BY ps.session_order ASC
|
|
""",
|
|
[program_id]
|
|
)
|
|
|
|
# Process sessions to extract exercise IDs and fetch exercise names
|
|
sessions_with_exercises = []
|
|
if sessions:
|
|
for session in sessions:
|
|
exercise_ids = []
|
|
if session.get('tag_filter'):
|
|
# Parse the filter string (e.g., "?exercise_id=5&exercise_id=1009")
|
|
parsed_filter = parse_qs(session['tag_filter'].lstrip('?'))
|
|
exercise_ids_str = parsed_filter.get('exercise_id', [])
|
|
try:
|
|
# Ensure IDs are unique and sorted if needed, though order might matter from filter
|
|
exercise_ids = sorted(list(set(int(eid) for eid in exercise_ids_str)))
|
|
except ValueError:
|
|
print(f"Warning: Could not parse exercise IDs from filter for tag {session['tag_id']}: {session['tag_filter']}")
|
|
exercise_ids = [] # Handle parsing error gracefully
|
|
|
|
exercises = []
|
|
if exercise_ids:
|
|
# Fetch exercise details for the extracted IDs
|
|
# Using tuple() for IN clause compatibility
|
|
# Ensure tuple has at least one element for SQL IN clause
|
|
if len(exercise_ids) == 1:
|
|
exercises_tuple = (exercise_ids[0],) # Comma makes it a tuple
|
|
else:
|
|
exercises_tuple = tuple(exercise_ids)
|
|
|
|
exercises = db.execute(
|
|
"SELECT exercise_id, name FROM exercise WHERE exercise_id IN %s ORDER BY name",
|
|
[exercises_tuple]
|
|
)
|
|
if exercises is None: exercises = [] # Ensure it's iterable
|
|
|
|
sessions_with_exercises.append({
|
|
**session, # Include all original session/tag data
|
|
'exercises': exercises
|
|
})
|
|
|
|
# Prepare context for the template
|
|
context = {
|
|
'program': program,
|
|
'sessions': sessions_with_exercises
|
|
}
|
|
|
|
# Check for HTMX request (optional, for potential future use)
|
|
if htmx:
|
|
# Assuming you have a block named 'content' in program_view.html
|
|
return render_block(current_app.jinja_env, 'program_view.html', 'content', **context)
|
|
else:
|
|
return render_template('program_view.html', **context)
|
|
|
|
# TODO: Add routes for editing and assigning programs |