Initial commit
This commit is contained in:
223
database.py
Normal file
223
database.py
Normal file
@@ -0,0 +1,223 @@
|
||||
import sqlite3
|
||||
import datetime
|
||||
from typing import List, Optional, Dict, Any
|
||||
|
||||
class Database:
|
||||
def __init__(self, db_name="exercise_diary.db"):
|
||||
self.db_name = db_name
|
||||
self.init_db()
|
||||
|
||||
def get_connection(self):
|
||||
return sqlite3.connect(self.db_name)
|
||||
|
||||
def init_db(self):
|
||||
"""Initialize the database tables."""
|
||||
conn = self.get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Daily Log Table (One entry per day)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS daily_logs (
|
||||
date TEXT PRIMARY KEY,
|
||||
weight REAL,
|
||||
waist REAL,
|
||||
sleep_hours REAL,
|
||||
sleep_quality INTEGER,
|
||||
mood INTEGER,
|
||||
notes TEXT
|
||||
)
|
||||
''')
|
||||
|
||||
# Exercises Table (Library of available exercises)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS exercises (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
description TEXT,
|
||||
type TEXT DEFAULT 'reps' -- 'reps' or 'time'
|
||||
)
|
||||
''')
|
||||
|
||||
# Pre-populate some basic calisthenics exercises if empty
|
||||
cursor.execute("SELECT count(*) FROM exercises")
|
||||
if cursor.fetchone()[0] == 0:
|
||||
basic_exercises = [
|
||||
("Pushups", "Standard pushups", "reps"),
|
||||
("Squats", "Bodyweight squats", "reps"),
|
||||
("Plank", "Static hold", "time"),
|
||||
("Lunges", "Alternating lunges", "reps"),
|
||||
("Crunches", "Abdominal crunches", "reps"),
|
||||
("Burpees", "Full body cardio/strength", "reps"),
|
||||
("Wall Sit", "Static leg hold", "time")
|
||||
]
|
||||
cursor.executemany("INSERT INTO exercises (name, description, type) VALUES (?, ?, ?)", basic_exercises)
|
||||
|
||||
# Workout Sessions Table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS workout_sessions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
date TEXT NOT NULL,
|
||||
notes TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
''')
|
||||
|
||||
# Workout Sets Table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS workout_sets (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
session_id INTEGER,
|
||||
exercise_id INTEGER,
|
||||
reps INTEGER,
|
||||
duration_seconds INTEGER,
|
||||
rpe INTEGER, -- 1-10 intensity
|
||||
variation TEXT, -- e.g., "Knee pushups"
|
||||
FOREIGN KEY (session_id) REFERENCES workout_sessions (id),
|
||||
FOREIGN KEY (exercise_id) REFERENCES exercises (id)
|
||||
)
|
||||
''')
|
||||
|
||||
# User Profile Table (Single row for settings)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS user_profile (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
height_cm REAL,
|
||||
goal_weight_kg REAL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
''')
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# --- User Profile Operations ---
|
||||
def get_user_profile(self) -> Optional[Dict]:
|
||||
conn = self.get_connection()
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT * FROM user_profile WHERE id = 1")
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
return dict(row) if row else None
|
||||
|
||||
def save_user_profile(self, height_cm: float, goal_weight_kg: float):
|
||||
conn = self.get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("INSERT OR REPLACE INTO user_profile (id, height_cm, goal_weight_kg) VALUES (1, ?, ?)", (height_cm, goal_weight_kg))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# --- Daily Log Operations ---
|
||||
def save_daily_log(self, date: str, data: Dict[str, Any]):
|
||||
"""Insert or update a daily log."""
|
||||
conn = self.get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Check if exists
|
||||
cursor.execute("SELECT date FROM daily_logs WHERE date = ?", (date,))
|
||||
exists = cursor.fetchone()
|
||||
|
||||
if exists:
|
||||
# Dynamic update based on keys provided
|
||||
set_clause = ", ".join([f"{key} = ?" for key in data.keys()])
|
||||
values = list(data.values()) + [date]
|
||||
cursor.execute(f"UPDATE daily_logs SET {set_clause} WHERE date = ?", values)
|
||||
else:
|
||||
columns = ", ".join(["date"] + list(data.keys()))
|
||||
placeholders = ", ".join(["?"] * (len(data) + 1))
|
||||
values = [date] + list(data.values())
|
||||
cursor.execute(f"INSERT INTO daily_logs ({columns}) VALUES ({placeholders})", values)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def get_daily_log(self, date: str) -> Optional[Dict]:
|
||||
conn = self.get_connection()
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT * FROM daily_logs WHERE date = ?", (date,))
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
if row:
|
||||
return dict(row)
|
||||
return None
|
||||
|
||||
def get_log_history(self, limit=30) -> List[Dict]:
|
||||
"""Get the last N days of logs."""
|
||||
conn = self.get_connection()
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT * FROM daily_logs ORDER BY date DESC LIMIT ?", (limit,))
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
# --- Exercise Operations ---
|
||||
def get_all_exercises(self) -> List[Dict]:
|
||||
conn = self.get_connection()
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT * FROM exercises ORDER BY name")
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
def add_exercise(self, name, description="", type="reps"):
|
||||
conn = self.get_connection()
|
||||
try:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("INSERT INTO exercises (name, description, type) VALUES (?, ?, ?)", (name, description, type))
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False # Name already exists
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# --- Workout Operations ---
|
||||
def create_workout_session(self, date: str, notes: str = "") -> int:
|
||||
conn = self.get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("INSERT INTO workout_sessions (date, notes) VALUES (?, ?)", (date, notes))
|
||||
session_id = cursor.lastrowid
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return session_id
|
||||
|
||||
def add_set(self, session_id: int, exercise_id: int, reps: int = 0, duration: int = 0, rpe: int = 0, variation: str = ""):
|
||||
conn = self.get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
INSERT INTO workout_sets (session_id, exercise_id, reps, duration_seconds, rpe, variation)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
''', (session_id, exercise_id, reps, duration, rpe, variation))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def delete_set(self, set_id: int):
|
||||
conn = self.get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM workout_sets WHERE id = ?", (set_id,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def get_workouts_by_date(self, date: str) -> List[Dict]:
|
||||
conn = self.get_connection()
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
# Get sessions
|
||||
cursor.execute("SELECT * FROM workout_sessions WHERE date = ?", (date,))
|
||||
sessions = [dict(row) for row in cursor.fetchall()]
|
||||
|
||||
# Get sets for each session
|
||||
for session in sessions:
|
||||
cursor.execute('''
|
||||
SELECT s.*, e.name as exercise_name
|
||||
FROM workout_sets s
|
||||
JOIN exercises e ON s.exercise_id = e.id
|
||||
WHERE s.session_id = ?
|
||||
''', (session['id'],))
|
||||
session['sets'] = [dict(row) for row in cursor.fetchall()]
|
||||
|
||||
conn.close()
|
||||
return sessions
|
||||
Reference in New Issue
Block a user