Initial commit
This commit is contained in:
@@ -0,0 +1,176 @@
|
||||
from datetime import date as date_type
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, status
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.access import product_barcode_accessible
|
||||
from app.deps import CurrentUserDep, SessionDep
|
||||
from app.models import FoodLog, Product
|
||||
from app.schemas import (
|
||||
DailyHistoryEntry,
|
||||
DailyLogResponse,
|
||||
DailyTotals,
|
||||
HistoryResponse,
|
||||
LogCreate,
|
||||
LogRead,
|
||||
LogUpdate,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/logs", tags=["logs"])
|
||||
|
||||
|
||||
def _day_bounds(day: date_type) -> tuple[datetime, datetime]:
|
||||
start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc)
|
||||
return start, start + timedelta(days=1)
|
||||
|
||||
|
||||
def _totals_for_logs(logs: list[FoodLog]) -> DailyTotals:
|
||||
return DailyTotals(
|
||||
calories=sum(log.calories for log in logs),
|
||||
carbs=sum(log.carbs for log in logs),
|
||||
protein=sum(log.protein for log in logs),
|
||||
fat=sum(log.fat for log in logs),
|
||||
sugar=sum(log.sugar for log in logs),
|
||||
fiber=sum(log.fiber for log in logs),
|
||||
saturated_fat=sum(log.saturated_fat for log in logs),
|
||||
salt=sum(log.salt for log in logs),
|
||||
)
|
||||
|
||||
|
||||
def _day_response(session: Session, user_id: int, day: date_type) -> DailyLogResponse:
|
||||
start, end = _day_bounds(day)
|
||||
logs = session.exec(
|
||||
select(FoodLog)
|
||||
.where(FoodLog.user_id == user_id)
|
||||
.where(FoodLog.timestamp >= start)
|
||||
.where(FoodLog.timestamp < end)
|
||||
.order_by(FoodLog.timestamp)
|
||||
).all()
|
||||
return DailyLogResponse(logs=logs, totals=_totals_for_logs(logs))
|
||||
|
||||
|
||||
@router.post("", response_model=LogRead, status_code=status.HTTP_201_CREATED)
|
||||
def create_log(log_in: LogCreate, current_user: CurrentUserDep, session: SessionDep):
|
||||
product = None
|
||||
if log_in.barcode:
|
||||
product = session.exec(select(Product).where(Product.barcode == log_in.barcode)).first()
|
||||
elif log_in.name:
|
||||
product = session.exec(select(Product).where(Product.name == log_in.name)).first()
|
||||
else:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="barcode oder name erforderlich")
|
||||
|
||||
if product is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Produkt nicht gefunden. Zuerst per GET /api/products/{barcode} suchen oder per POST /api/products anlegen.",
|
||||
)
|
||||
|
||||
if not product_barcode_accessible(session, product.barcode, current_user.id):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Produkt nicht gefunden. Zuerst per GET /api/products/{barcode} suchen oder per POST /api/products anlegen.",
|
||||
)
|
||||
|
||||
factor = log_in.amount_g / 100.0
|
||||
log = FoodLog(
|
||||
user_id=current_user.id,
|
||||
barcode=product.barcode,
|
||||
name=product.name,
|
||||
meal_type=log_in.meal_type,
|
||||
amount_g=log_in.amount_g,
|
||||
calories=product.calories * factor,
|
||||
carbs=product.carbs * factor,
|
||||
protein=product.protein * factor,
|
||||
fat=product.fat * factor,
|
||||
sugar=product.sugar * factor,
|
||||
fiber=product.fiber * factor,
|
||||
saturated_fat=product.saturated_fat * factor,
|
||||
salt=product.salt * factor,
|
||||
)
|
||||
session.add(log)
|
||||
session.commit()
|
||||
session.refresh(log)
|
||||
return log
|
||||
|
||||
|
||||
def _get_owned_log(session: Session, log_id: int, user_id: int) -> FoodLog:
|
||||
log = session.get(FoodLog, log_id)
|
||||
if log is None or log.user_id != user_id:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Eintrag nicht gefunden")
|
||||
return log
|
||||
|
||||
|
||||
@router.put("/{log_id}", response_model=LogRead)
|
||||
def update_log(log_id: int, log_in: LogUpdate, current_user: CurrentUserDep, session: SessionDep):
|
||||
log = _get_owned_log(session, log_id, current_user.id)
|
||||
|
||||
if log_in.amount_g is not None and log_in.amount_g != log.amount_g:
|
||||
factor = log_in.amount_g / log.amount_g
|
||||
log.calories *= factor
|
||||
log.carbs *= factor
|
||||
log.protein *= factor
|
||||
log.fat *= factor
|
||||
log.sugar *= factor
|
||||
log.fiber *= factor
|
||||
log.saturated_fat *= factor
|
||||
log.salt *= factor
|
||||
log.amount_g = log_in.amount_g
|
||||
|
||||
if log_in.meal_type is not None:
|
||||
log.meal_type = log_in.meal_type
|
||||
|
||||
session.add(log)
|
||||
session.commit()
|
||||
session.refresh(log)
|
||||
return log
|
||||
|
||||
|
||||
@router.delete("/{log_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_log(log_id: int, current_user: CurrentUserDep, session: SessionDep):
|
||||
log = _get_owned_log(session, log_id, current_user.id)
|
||||
session.delete(log)
|
||||
session.commit()
|
||||
|
||||
|
||||
@router.get("/today", response_model=DailyLogResponse)
|
||||
def get_today_logs(current_user: CurrentUserDep, session: SessionDep):
|
||||
today = datetime.now(timezone.utc).date()
|
||||
return _day_response(session, current_user.id, today)
|
||||
|
||||
|
||||
@router.get("/history", response_model=HistoryResponse)
|
||||
def get_history(
|
||||
current_user: CurrentUserDep,
|
||||
session: SessionDep,
|
||||
days: int = Query(default=30, ge=1, le=365),
|
||||
):
|
||||
today = datetime.now(timezone.utc).date()
|
||||
range_start_day = today - timedelta(days=days - 1)
|
||||
range_start, _ = _day_bounds(range_start_day)
|
||||
_, range_end = _day_bounds(today)
|
||||
|
||||
logs = session.exec(
|
||||
select(FoodLog)
|
||||
.where(FoodLog.user_id == current_user.id)
|
||||
.where(FoodLog.timestamp >= range_start)
|
||||
.where(FoodLog.timestamp < range_end)
|
||||
).all()
|
||||
|
||||
logs_by_day: dict[date_type, list[FoodLog]] = {}
|
||||
for log in logs:
|
||||
logs_by_day.setdefault(log.timestamp.date(), []).append(log)
|
||||
|
||||
result_days = [
|
||||
DailyHistoryEntry(
|
||||
date=range_start_day + timedelta(days=offset),
|
||||
totals=_totals_for_logs(logs_by_day.get(range_start_day + timedelta(days=offset), [])),
|
||||
)
|
||||
for offset in range(days)
|
||||
]
|
||||
return HistoryResponse(days=result_days)
|
||||
|
||||
|
||||
@router.get("/day/{day}", response_model=DailyLogResponse)
|
||||
def get_day_logs(day: date_type, current_user: CurrentUserDep, session: SessionDep):
|
||||
return _day_response(session, current_user.id, day)
|
||||
Reference in New Issue
Block a user