Lars
cbcb6a2a34
- Weight, Circumference, Caliper now BLOCK on limit exceeded - Raise HTTPException(403) with user-friendly message - Show used/limit and suggest contacting admin - Phase 2 → Phase 4 transition Phase 4: Enforcement (Batch 1/3) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
101 lines
4.0 KiB
Python
101 lines
4.0 KiB
Python
"""
|
|
Circumference Tracking Endpoints for Mitai Jinkendo
|
|
|
|
Handles body circumference measurements (8 measurement points).
|
|
"""
|
|
import uuid
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Header, Depends, HTTPException
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from auth import require_auth, check_feature_access, increment_feature_usage
|
|
from models import CircumferenceEntry
|
|
from routers.profiles import get_pid
|
|
from feature_logger import log_feature_usage
|
|
|
|
router = APIRouter(prefix="/api/circumferences", tags=["circumference"])
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@router.get("")
|
|
def list_circs(limit: int=100, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
|
"""Get circumference entries for current profile."""
|
|
pid = get_pid(x_profile_id)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"SELECT * FROM circumference_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_circ(e: CircumferenceEntry, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
|
"""Create or update circumference entry (upsert by date)."""
|
|
pid = get_pid(x_profile_id)
|
|
|
|
# Phase 4: Check feature access and ENFORCE
|
|
access = check_feature_access(pid, 'circumference_entries')
|
|
log_feature_usage(pid, 'circumference_entries', access, 'create')
|
|
|
|
if not access['allowed']:
|
|
logger.warning(
|
|
f"[FEATURE-LIMIT] User {pid} blocked: "
|
|
f"circumference_entries {access['reason']} (used: {access['used']}, limit: {access['limit']})"
|
|
)
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Limit erreicht: Du hast das Kontingent für Umfangs-Einträge überschritten ({access['used']}/{access['limit']}). "
|
|
f"Bitte kontaktiere den Admin oder warte bis zum nächsten Reset."
|
|
)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("SELECT id FROM circumference_log WHERE profile_id=%s AND date=%s", (pid,e.date))
|
|
ex = cur.fetchone()
|
|
d = e.model_dump()
|
|
is_new_entry = not ex
|
|
|
|
if ex:
|
|
# UPDATE existing entry
|
|
eid = ex['id']
|
|
sets = ', '.join(f"{k}=%s" for k in d if k!='date')
|
|
cur.execute(f"UPDATE circumference_log SET {sets} WHERE id=%s",
|
|
[v for k,v in d.items() if k!='date']+[eid])
|
|
else:
|
|
# INSERT new entry
|
|
eid = str(uuid.uuid4())
|
|
cur.execute("""INSERT INTO circumference_log
|
|
(id,profile_id,date,c_neck,c_chest,c_waist,c_belly,c_hip,c_thigh,c_calf,c_arm,notes,photo_id,created)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,CURRENT_TIMESTAMP)""",
|
|
(eid,pid,d['date'],d['c_neck'],d['c_chest'],d['c_waist'],d['c_belly'],
|
|
d['c_hip'],d['c_thigh'],d['c_calf'],d['c_arm'],d['notes'],d['photo_id']))
|
|
|
|
# Phase 2: Increment usage counter (only for new entries)
|
|
increment_feature_usage(pid, 'circumference_entries')
|
|
|
|
return {"id":eid,"date":e.date}
|
|
|
|
|
|
@router.put("/{eid}")
|
|
def update_circ(eid: str, e: CircumferenceEntry, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
|
"""Update existing circumference entry."""
|
|
pid = get_pid(x_profile_id)
|
|
with get_db() as conn:
|
|
d = e.model_dump()
|
|
cur = get_cursor(conn)
|
|
cur.execute(f"UPDATE circumference_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_circ(eid: str, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
|
|
"""Delete circumference entry."""
|
|
pid = get_pid(x_profile_id)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("DELETE FROM circumference_log WHERE id=%s AND profile_id=%s", (eid,pid))
|
|
return {"ok":True}
|