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
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from typing import List
from database import get_db
from models import Ingredient
from schemas import IngredientCreate, IngredientRead, IngredientUpdate
router = APIRouter(prefix="/api/pantry", tags=["pantry"])
@router.get("", response_model=List[IngredientRead])
def list_ingredients(
expiring_soon: bool = Query(False),
db: Session = Depends(get_db)
):
"""List all ingredients. Optional query param ?expiring_soon=true filters to ingredients with expiry_date within 7 days from today."""
query = db.query(Ingredient)
if expiring_soon:
today = datetime.utcnow().date()
seven_days_later = today + timedelta(days=7)
query = query.filter(
(Ingredient.expiry_date >= today) & (Ingredient.expiry_date <= seven_days_later)
)
return query.all()
@router.get("/expiring", response_model=List[IngredientRead])
def list_expiring_ingredients(db: Session = Depends(get_db)):
"""Ingredients expiring within 7 days."""
today = datetime.utcnow().date()
seven_days_later = today + timedelta(days=7)
return db.query(Ingredient).filter(
(Ingredient.expiry_date >= today) & (Ingredient.expiry_date <= seven_days_later)
).all()
@router.post("", response_model=IngredientRead, status_code=status.HTTP_201_CREATED)
def create_ingredient(ingredient: IngredientCreate, db: Session = Depends(get_db)):
"""Create a new ingredient."""
db_ingredient = Ingredient(**ingredient.model_dump())
db.add(db_ingredient)
db.commit()
db.refresh(db_ingredient)
return db_ingredient
@router.put("/{id}", response_model=IngredientRead)
def update_ingredient(id: int, ingredient: IngredientUpdate, db: Session = Depends(get_db)):
"""Update ingredient fields (partial update)."""
db_ingredient = db.query(Ingredient).filter(Ingredient.id == id).first()
if not db_ingredient:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient not found")
update_data = ingredient.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_ingredient, field, value)
db_ingredient.updated_at = datetime.utcnow()
db.commit()
db.refresh(db_ingredient)
return db_ingredient
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_ingredient(id: int, db: Session = Depends(get_db)):
"""Delete an ingredient."""
db_ingredient = db.query(Ingredient).filter(Ingredient.id == id).first()
if not db_ingredient:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient not found")
db.delete(db_ingredient)
db.commit()
@router.post("/bulk", response_model=List[IngredientRead])
def bulk_upsert_ingredients(ingredients: List[IngredientCreate], db: Session = Depends(get_db)):
"""Upsert list of ingredients. If an ingredient with same name exists, add the quantity (accumulate). If not, create new."""
result = []
for ingredient_data in ingredients:
existing = db.query(Ingredient).filter(Ingredient.name == ingredient_data.name).first()
if existing:
existing.quantity += ingredient_data.quantity
existing.updated_at = datetime.utcnow()
db.commit()
db.refresh(existing)
result.append(existing)
else:
new_ingredient = Ingredient(**ingredient_data.model_dump())
db.add(new_ingredient)
db.commit()
db.refresh(new_ingredient)
result.append(new_ingredient)
return result
|
