76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
|
"""
|
||
|
|
Caliper/Skinfold Tracking Endpoints for Mitai Jinkendo
|
||
|
|
|
||
|
|
Handles body fat measurements via skinfold caliper (4 methods supported).
|
||
|
|
"""
|
||
|
|
import uuid
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Header, Depends
|
||
|
|
|
||
|
|
from db import get_db, get_cursor, r2d
|
||
|
|
from auth import require_auth
|
||
|
|
from models import CaliperEntry
|
||
|
|
from routers.profiles import get_pid
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/caliper", tags=["caliper"])
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("")
|
||
|
|
def list_caliper(limit: int=100, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
||
|
|
"""Get caliper entries for current profile."""
|
||
|
|
pid = get_pid(x_profile_id)
|
||
|
|
with get_db() as conn:
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute(
|
||
|
|
"SELECT * FROM caliper_log WHERE profile_id=%s ORDER BY date DESC LIMIT %s", (pid,limit))
|
||
|
|
return [r2d(r) for r in cur.fetchall()]
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("")
|
||
|
|
def upsert_caliper(e: CaliperEntry, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
||
|
|
"""Create or update caliper entry (upsert by date)."""
|
||
|
|
pid = get_pid(x_profile_id)
|
||
|
|
with get_db() as conn:
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute("SELECT id FROM caliper_log WHERE profile_id=%s AND date=%s", (pid,e.date))
|
||
|
|
ex = cur.fetchone()
|
||
|
|
d = e.model_dump()
|
||
|
|
if ex:
|
||
|
|
eid = ex['id']
|
||
|
|
sets = ', '.join(f"{k}=%s" for k in d if k!='date')
|
||
|
|
cur.execute(f"UPDATE caliper_log SET {sets} WHERE id=%s",
|
||
|
|
[v for k,v in d.items() if k!='date']+[eid])
|
||
|
|
else:
|
||
|
|
eid = str(uuid.uuid4())
|
||
|
|
cur.execute("""INSERT INTO caliper_log
|
||
|
|
(id,profile_id,date,sf_method,sf_chest,sf_axilla,sf_triceps,sf_subscap,sf_suprailiac,
|
||
|
|
sf_abdomen,sf_thigh,sf_calf_med,sf_lowerback,sf_biceps,body_fat_pct,lean_mass,fat_mass,notes,created)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,CURRENT_TIMESTAMP)""",
|
||
|
|
(eid,pid,d['date'],d['sf_method'],d['sf_chest'],d['sf_axilla'],d['sf_triceps'],
|
||
|
|
d['sf_subscap'],d['sf_suprailiac'],d['sf_abdomen'],d['sf_thigh'],d['sf_calf_med'],
|
||
|
|
d['sf_lowerback'],d['sf_biceps'],d['body_fat_pct'],d['lean_mass'],d['fat_mass'],d['notes']))
|
||
|
|
return {"id":eid,"date":e.date}
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/{eid}")
|
||
|
|
def update_caliper(eid: str, e: CaliperEntry, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
||
|
|
"""Update existing caliper entry."""
|
||
|
|
pid = get_pid(x_profile_id)
|
||
|
|
with get_db() as conn:
|
||
|
|
d = e.model_dump()
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute(f"UPDATE caliper_log SET {', '.join(f'{k}=%s' for k in d)} WHERE id=%s AND profile_id=%s",
|
||
|
|
list(d.values())+[eid,pid])
|
||
|
|
return {"id":eid}
|
||
|
|
|
||
|
|
|
||
|
|
@router.delete("/{eid}")
|
||
|
|
def delete_caliper(eid: str, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
||
|
|
"""Delete caliper entry."""
|
||
|
|
pid = get_pid(x_profile_id)
|
||
|
|
with get_db() as conn:
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute("DELETE FROM caliper_log WHERE id=%s AND profile_id=%s", (eid,pid))
|
||
|
|
return {"ok":True}
|