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
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from typing import List, Dict, Any
from collections import defaultdict
from database import get_db
from models import MealLog, MealIngredient
from schemas import MealLogCreate, MealLogRead, MealIngredientCreate
router = APIRouter(prefix="/api/meals", tags=["meals"])
@router.get("", response_model=List[MealLogRead])
def list_meals(
days: int = Query(30),
meal_type: str = Query(None),
db: Session = Depends(get_db)
):
"""List meal log entries. Optional query params: ?days=30 (default 30, filters entries from last N days), ?meal_type=dinner. Returns entries ordered by eaten_at desc."""
cutoff_date = datetime.utcnow() - timedelta(days=days)
query = db.query(MealLog).filter(MealLog.eaten_at >= cutoff_date)
if meal_type:
query = query.filter(MealLog.meal_type == meal_type)
return query.order_by(MealLog.eaten_at.desc()).all()
@router.get("/stats", response_model=Dict[str, Any])
def meal_stats(db: Session = Depends(get_db)):
"""Return meal frequency stats. Query last 30 days, count occurrences of each meal_name. Return {"meal_frequency": {"Pasta Carbonara": 3, ...}, "total_meals": 15, "by_type": {"dinner": 10, "lunch": 5}}."""
cutoff_date = datetime.utcnow() - timedelta(days=30)
meals = db.query(MealLog).filter(MealLog.eaten_at >= cutoff_date).all()
meal_frequency = defaultdict(int)
meal_by_type = defaultdict(int)
for meal in meals:
meal_frequency[meal.meal_name] += 1
meal_by_type[meal.meal_type] += 1
return {
"meal_frequency": dict(meal_frequency),
"total_meals": len(meals),
"by_type": dict(meal_by_type)
}
@router.get("/{id}", response_model=MealLogRead)
def get_meal(id: int, db: Session = Depends(get_db)):
"""Get a single meal log entry."""
meal = db.query(MealLog).filter(MealLog.id == id).first()
if not meal:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Meal not found")
return meal
@router.post("", response_model=MealLogRead, status_code=status.HTTP_201_CREATED)
def create_meal(meal: MealLogCreate, db: Session = Depends(get_db)):
"""Log a new meal. If ingredients list provided, create MealIngredient records linked to this meal."""
# Create the MealLog entry
meal_data = meal.model_dump(exclude={"ingredients"})
if meal_data["eaten_at"] is None:
meal_data["eaten_at"] = datetime.utcnow()
db_meal = MealLog(**meal_data)
db.add(db_meal)
db.commit()
db.refresh(db_meal)
# Create MealIngredient entries if provided
if meal.ingredients:
for ingredient in meal.ingredients:
meal_ingredient = MealIngredient(
meal_log_id=db_meal.id,
ingredient_name=ingredient.ingredient_name,
quantity_used=ingredient.quantity_used,
unit=ingredient.unit
)
db.add(meal_ingredient)
db.commit()
db.refresh(db_meal)
return db_meal
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_meal(id: int, db: Session = Depends(get_db)):
"""Delete a meal log entry."""
meal = db.query(MealLog).filter(MealLog.id == id).first()
if not meal:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Meal not found")
db.delete(meal)
db.commit()
|
