1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
import sqlite3
def add_exercise(conn: sqlite3.Connection, name: str, note: str | None = None) -> int:
cur = conn.execute("INSERT INTO exercises (name, note) VALUES (?, ?)", (name, note))
conn.commit()
assert cur.lastrowid is not None
return cur.lastrowid
def list_exercises(conn: sqlite3.Connection) -> list[sqlite3.Row]:
return conn.execute("SELECT id, name, note FROM exercises ORDER BY name").fetchall()
def get_exercise(conn: sqlite3.Connection, exercise_id: int) -> sqlite3.Row | None:
return conn.execute(
"SELECT id, name, note FROM exercises WHERE id = ?", (exercise_id,)
).fetchone()
def update_exercise(
conn: sqlite3.Connection,
exercise_id: int,
name: str | None = None,
note: str | None = None,
) -> bool:
ex = get_exercise(conn, exercise_id)
if not ex:
return False
conn.execute(
"UPDATE exercises SET name = ?, note = ? WHERE id = ?",
(name or ex["name"], note, exercise_id),
)
conn.commit()
return True
def delete_exercise(conn: sqlite3.Connection, exercise_id: int) -> None:
conn.execute("DELETE FROM exercises WHERE id = ?", (exercise_id,))
conn.commit()
def create_session(
conn: sqlite3.Connection, date_time: str, note: str | None = None
) -> int:
cur = conn.execute(
"INSERT INTO sessions (date_time, note) VALUES (?, ?)", (date_time, note)
)
assert cur.lastrowid is not None
return cur.lastrowid
def list_sessions(conn: sqlite3.Connection, limit: int = 20) -> list[sqlite3.Row]:
return conn.execute(
"""SELECT s.id, s.date_time, s.note,
GROUP_CONCAT(e.name, ', ') as exercises
FROM sessions s
LEFT JOIN session_exercises se ON s.id = se.session_id
LEFT JOIN exercises e ON se.exercise_id = e.id
GROUP BY s.id
ORDER BY s.date_time DESC
LIMIT ?""",
(limit,),
).fetchall()
def get_session_detail(
conn: sqlite3.Connection, session_id: int
) -> tuple[sqlite3.Row | None, list[sqlite3.Row]]:
session = conn.execute(
"SELECT id, date_time, note FROM sessions WHERE id = ?", (session_id,)
).fetchone()
if not session:
return None, []
entries = conn.execute(
"""SELECT se.id, se.position, e.name as exercise_name,
se.sets, se.reps, se.rpe, se.rest_time, se.lsrpe, se.note
FROM session_exercises se
JOIN exercises e ON se.exercise_id = e.id
WHERE se.session_id = ?
ORDER BY se.position""",
(session_id,),
).fetchall()
return session, entries
def delete_session(conn: sqlite3.Connection, session_id: int) -> None:
conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
conn.commit()
def add_session_exercise(
conn: sqlite3.Connection,
session_id: int,
exercise_id: int,
position: int,
sets: int,
reps: int,
rpe: int | None = None,
rest_time: int | None = None,
lsrpe: int | None = None,
note: str | None = None,
) -> int:
cur = conn.execute(
"""INSERT INTO session_exercises
(session_id, exercise_id, position, sets, reps, rpe, rest_time, lsrpe, note)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(session_id, exercise_id, position, sets, reps, rpe, rest_time, lsrpe, note),
)
assert cur.lastrowid is not None
return cur.lastrowid
def save_session(
conn: sqlite3.Connection, date_time: str, note: str | None, exercises: list[dict]
) -> int:
session_id = create_session(conn, date_time, note)
for i, ex in enumerate(exercises, 1):
add_session_exercise(
conn,
session_id,
ex["exercise_id"],
i,
ex["sets"],
ex["reps"],
ex.get("rpe"),
ex.get("rest_time"),
ex.get("lsrpe"),
ex.get("note"),
)
conn.commit()
return session_id
|
