Switch to using Postgres database instead of sqlite
This commit is contained in:
Peter Stockings
159
db.py
159
db.py
@@ -1,5 +1,8 @@
|
||||
import os
|
||||
import psycopg2
|
||||
from psycopg2.extras import RealDictCursor
|
||||
import datetime
|
||||
import sqlite3
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from utils import get_all_exercises_from_topsets, get_people_and_exercise_rep_maxes, get_workouts
|
||||
|
||||
@@ -8,100 +11,116 @@ class DataBase():
|
||||
def __init__(self, app):
|
||||
self.DATABASE_URI = app.config['DATABASE_URI']
|
||||
|
||||
db_url = urlparse(os.environ['DATABASE_URL'])
|
||||
username = db_url.username
|
||||
password = db_url.password
|
||||
database = db_url.path[1:]
|
||||
hostname = db_url.hostname
|
||||
port = db_url.port
|
||||
|
||||
self.conn = psycopg2.connect(
|
||||
database=database,
|
||||
user=username,
|
||||
password=password,
|
||||
host=hostname,
|
||||
port=port
|
||||
)
|
||||
|
||||
def execute(self, query, args=(), one=False, commit=False):
|
||||
conn = sqlite3.connect(self.DATABASE_URI)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cur = conn.execute(query, args)
|
||||
cur = self.conn.cursor(cursor_factory=RealDictCursor)
|
||||
cur.execute(query, args)
|
||||
rv = None
|
||||
if cur.description is not None:
|
||||
rv = cur.fetchall()
|
||||
if commit:
|
||||
conn.commit()
|
||||
self.conn.commit()
|
||||
cur.close()
|
||||
|
||||
return (rv[0] if rv else None) if one else rv
|
||||
|
||||
def get_exercises(self):
|
||||
exercises = self.execute('SELECT * FROM Exercise')
|
||||
exercises = self.execute(
|
||||
'SELECT ExerciseId AS "ExerciseId", Name AS "Name" FROM Exercise')
|
||||
return [{"ExerciseId": e['ExerciseId'], "Name": e['Name']} for e in exercises]
|
||||
|
||||
def get_person(self, person_id):
|
||||
person = self.execute(
|
||||
'SELECT * FROM Person WHERE PersonId=? LIMIT 1', [person_id], one=True)
|
||||
'SELECT PersonId AS "PersonId" FROM Person WHERE PersonId=%s LIMIT 1', [person_id], one=True)
|
||||
return person
|
||||
|
||||
def get_workout(self, person_id, workout_id):
|
||||
workout = self.execute('SELECT W.WorkoutId FROM Person P, Workout W WHERE P.PersonId=W.PersonId AND P.PersonId=? AND W.WorkoutId=? LIMIT 1', [
|
||||
workout = self.execute('SELECT W.WorkoutId AS "WorkoutId" FROM Person P, Workout W WHERE P.PersonId=W.PersonId AND P.PersonId=%s AND W.WorkoutId=%s LIMIT 1', [
|
||||
person_id, workout_id], one=True)
|
||||
return workout
|
||||
|
||||
def get_topset(self, person_id, workout_id, topset_id):
|
||||
topset = self.execute("""
|
||||
SELECT T.TopSetId
|
||||
SELECT T.TopSetId AS "TopSetId"
|
||||
FROM Person P, Workout W, TopSet T
|
||||
WHERE W.PersonId=W.PersonId AND W.WorkoutId=T.WorkoutId AND P.PersonId=? AND W.WorkoutId = ? AND T.TopSetId = ?
|
||||
WHERE W.PersonId=W.PersonId AND W.WorkoutId=T.WorkoutId AND P.PersonId=%s AND W.WorkoutId = %s AND T.TopSetId = %s
|
||||
LIMIT 1""", [person_id, workout_id, topset_id], one=True)
|
||||
return topset
|
||||
|
||||
def delete_workout(self, workout_id):
|
||||
self.execute('DELETE FROM TopSet WHERE WorkoutId=?',
|
||||
self.execute('DELETE FROM TopSet WHERE WorkoutId=%s',
|
||||
[workout_id], commit=True)
|
||||
self.execute('DELETE FROM Workout WHERE WorkoutId=?',
|
||||
self.execute('DELETE FROM Workout WHERE WorkoutId=%s',
|
||||
[workout_id], commit=True)
|
||||
|
||||
def update_topset(self, exercise_id, repetitions, weight, topset_id):
|
||||
self.execute('UPDATE TopSet SET ExerciseId=?, Repetitions=?, Weight=? WHERE TopSetId=?', [
|
||||
self.execute('UPDATE TopSet SET ExerciseId=%s, Repetitions=%s, Weight=%s WHERE TopSetId=%s', [
|
||||
exercise_id, repetitions, weight, topset_id], commit=True)
|
||||
|
||||
def create_topset(self, workout_id, exercise_id, repetitions, weight):
|
||||
self.execute('INSERT INTO TopSet (WorkoutId, ExerciseId, Repetitions, Weight) VALUES (?, ?, ?, ?)', [
|
||||
self.execute('INSERT INTO TopSet (WorkoutId, ExerciseId, Repetitions, Weight) VALUES (%s, %s, %s, %s)', [
|
||||
workout_id, exercise_id, repetitions, weight], commit=True)
|
||||
|
||||
def delete_topset(self, topset_id):
|
||||
self.execute('DELETE FROM TopSet WHERE TopSetId=?', [
|
||||
self.execute('DELETE FROM TopSet WHERE TopSetId=%s', [
|
||||
topset_id], commit=True)
|
||||
|
||||
def create_workout(self, person_id):
|
||||
now = datetime.datetime.now()
|
||||
date_string = now.strftime('%Y-%m-%d')
|
||||
print(f'Creating workout for {person_id} at {date_string}')
|
||||
self.execute('INSERT INTO Workout (PersonId, StartDate) VALUES (?, ?)', [
|
||||
person_id, date_string], commit=True)
|
||||
w = self.execute('SELECT MAX(WorkoutId) AS WorkoutId FROM Workout WHERE PersonId=?', [
|
||||
person_id], one=True)
|
||||
return w['WorkoutId']
|
||||
print(
|
||||
f'Creating workout for PersonId {person_id} starting at {date_string}')
|
||||
new_workout = self.execute('INSERT INTO Workout (PersonId, StartDate) VALUES (%s, %s) RETURNING WorkoutId AS "WorkoutId"', [
|
||||
person_id, date_string], commit=True, one=True)
|
||||
return new_workout['WorkoutId']
|
||||
|
||||
def get_people_and_workout_count(self, person_id):
|
||||
return self.execute("""
|
||||
SELECT
|
||||
P.PersonId,
|
||||
P.Name,
|
||||
COUNT(W.WorkoutId) AS NumberOfWorkouts,
|
||||
P.PersonId AS "PersonId",
|
||||
P.Name AS "Name",
|
||||
COUNT(W.WorkoutId) AS "NumberOfWorkouts",
|
||||
CASE P.PersonId
|
||||
WHEN ?
|
||||
WHEN %s
|
||||
THEN 1
|
||||
ELSE 0
|
||||
END IsActive
|
||||
END "IsActive"
|
||||
FROM
|
||||
Person P
|
||||
LEFT JOIN Workout W ON P.PersonId = W.PersonId
|
||||
Person P LEFT JOIN Workout W ON P.PersonId = W.PersonId
|
||||
GROUP BY
|
||||
P.PersonId""", [person_id])
|
||||
|
||||
def get_person_final(self, person_id):
|
||||
topsets = self.execute("""
|
||||
SELECT
|
||||
P.PersonId,
|
||||
P.Name AS PersonName,
|
||||
W.WorkoutId,
|
||||
W.StartDate,
|
||||
T.TopSetId,
|
||||
E.ExerciseId,
|
||||
E.Name AS ExerciseName,
|
||||
T.Repetitions,
|
||||
T.Weight
|
||||
P.PersonId AS "PersonId",
|
||||
P.Name AS "PersonName",
|
||||
W.WorkoutId AS "WorkoutId",
|
||||
W.StartDate AS "StartDate",
|
||||
T.TopSetId AS "TopSetId",
|
||||
E.ExerciseId AS "ExerciseId",
|
||||
E.Name AS "ExerciseName",
|
||||
T.Repetitions AS "Repetitions",
|
||||
T.Weight AS "Weight"
|
||||
FROM Person P
|
||||
LEFT JOIN Workout W ON P.PersonId=W.PersonId
|
||||
LEFT JOIN TopSet T ON W.WorkoutId=T.WorkoutId
|
||||
LEFT JOIN Exercise E ON T.ExerciseId=E.ExerciseId
|
||||
WHERE P.PersonId=?""", [person_id])
|
||||
WHERE P.PersonId=%s""", [person_id])
|
||||
|
||||
return {
|
||||
'PersonId': next((t['PersonId'] for t in topsets), -1),
|
||||
@@ -113,21 +132,21 @@ class DataBase():
|
||||
def get_workout_final(self, person_id, workout_id):
|
||||
topsets = self.execute("""
|
||||
SELECT
|
||||
P.PersonId,
|
||||
P.Name AS PersonName,
|
||||
W.WorkoutId,
|
||||
W.StartDate,
|
||||
T.TopSetId,
|
||||
E.ExerciseId,
|
||||
E.Name AS ExerciseName,
|
||||
T.Repetitions,
|
||||
T.Weight
|
||||
P.PersonId AS "PersonId",
|
||||
P.Name AS "PersonName",
|
||||
W.WorkoutId AS "WorkoutId",
|
||||
W.StartDate AS "StartDate",
|
||||
T.TopSetId AS "TopSetId",
|
||||
E.ExerciseId AS "ExerciseId",
|
||||
E.Name AS "ExerciseName",
|
||||
T.Repetitions AS "Repetitions",
|
||||
T.Weight AS "Weight"
|
||||
FROM Person P
|
||||
LEFT JOIN Workout W ON P.PersonId=W.PersonId
|
||||
LEFT JOIN TopSet T ON W.WorkoutId=T.WorkoutId
|
||||
LEFT JOIN Exercise E ON T.ExerciseId=E.ExerciseId
|
||||
WHERE P.PersonId=?
|
||||
AND W.WorkoutId = ?""", [person_id, workout_id])
|
||||
WHERE P.PersonId=%s
|
||||
AND W.WorkoutId = %s""", [person_id, workout_id])
|
||||
|
||||
return {
|
||||
'PersonId': next((t['PersonId'] for t in topsets), -1),
|
||||
@@ -141,22 +160,22 @@ class DataBase():
|
||||
def get_topset_final(self, person_id, workout_id, topset_id):
|
||||
topset = self.execute("""
|
||||
SELECT
|
||||
P.PersonId,
|
||||
P.Name AS PersonName,
|
||||
W.WorkoutId,
|
||||
W.StartDate,
|
||||
T.TopSetId,
|
||||
E.ExerciseId,
|
||||
E.Name AS ExerciseName,
|
||||
T.Repetitions,
|
||||
T.Weight
|
||||
P.PersonId AS "PersonId",
|
||||
P.Name AS "PersonName",
|
||||
W.WorkoutId AS "WorkoutId",
|
||||
W.StartDate AS "StartDate",
|
||||
T.TopSetId AS "TopSetId",
|
||||
E.ExerciseId AS "ExerciseId",
|
||||
E.Name AS "ExerciseName",
|
||||
T.Repetitions AS "Repetitions",
|
||||
T.Weight AS "Weight"
|
||||
FROM Person P
|
||||
INNER JOIN Workout W ON P.PersonId=W.PersonId
|
||||
INNER JOIN TopSet T ON W.WorkoutId=T.WorkoutId
|
||||
INNER JOIN Exercise E ON T.ExerciseId=E.ExerciseId
|
||||
WHERE P.PersonId=?
|
||||
AND W.WorkoutId = ?
|
||||
AND T.TopSetId = ?""", [person_id, workout_id, topset_id], one=True)
|
||||
WHERE P.PersonId=%s
|
||||
AND W.WorkoutId = %s
|
||||
AND T.TopSetId = %s""", [person_id, workout_id, topset_id], one=True)
|
||||
|
||||
return {
|
||||
'PersonId': topset['PersonId'],
|
||||
@@ -174,15 +193,15 @@ class DataBase():
|
||||
def get_all_topsets(self):
|
||||
all_topsets = self.execute("""
|
||||
SELECT
|
||||
P.PersonId,
|
||||
P.Name AS PersonName,
|
||||
W.WorkoutId,
|
||||
W.StartDate,
|
||||
T.TopSetId,
|
||||
E.ExerciseId,
|
||||
E.Name AS ExerciseName,
|
||||
T.Repetitions,
|
||||
T.Weight
|
||||
P.PersonId AS "PersonId",
|
||||
P.Name AS "PersonName",
|
||||
W.WorkoutId AS "WorkoutId",
|
||||
W.StartDate AS "StartDate",
|
||||
T.TopSetId AS "TopSetId",
|
||||
E.ExerciseId AS "ExerciseId",
|
||||
E.Name AS "ExerciseName",
|
||||
T.Repetitions AS "Repetitions",
|
||||
T.Weight AS "Weight"
|
||||
FROM Person P
|
||||
LEFT JOIN Workout W ON P.PersonId=W.PersonId
|
||||
LEFT JOIN TopSet T ON W.WorkoutId=T.WorkoutId
|
||||
|
||||
@@ -2,3 +2,4 @@ Flask==2.0.1
|
||||
gunicorn==19.7.1
|
||||
Jinja2==3.0.1
|
||||
flasgger==0.9.5
|
||||
psycopg2-binary=2.9.3
|
||||
Reference in New Issue
Block a user