289 lines
11 KiB
Python
289 lines
11 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import text
|
|
from typing import List
|
|
import models, schemas
|
|
from database import engine, get_db
|
|
|
|
# Create database tables
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI(
|
|
title="Grocery Tracker API",
|
|
description="API for tracking grocery prices and shopping events",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# CORS middleware for React frontend
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["http://localhost:3000", "http://localhost:5173"], # React dev servers
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
def build_shopping_event_response(event: models.ShoppingEvent, db: Session) -> schemas.ShoppingEventResponse:
|
|
"""Build a shopping event response with groceries from the association table"""
|
|
# Get groceries with their event-specific data
|
|
grocery_data = db.execute(
|
|
text("""
|
|
SELECT g.id, g.name, g.category, g.organic, g.weight, g.weight_unit,
|
|
seg.amount, seg.price
|
|
FROM groceries g
|
|
JOIN shopping_event_groceries seg ON g.id = seg.grocery_id
|
|
WHERE seg.shopping_event_id = :event_id
|
|
"""),
|
|
{"event_id": event.id}
|
|
).fetchall()
|
|
|
|
# Convert to GroceryWithEventData objects
|
|
groceries_with_data = [
|
|
schemas.GroceryWithEventData(
|
|
id=row.id,
|
|
name=row.name,
|
|
category=row.category,
|
|
organic=row.organic,
|
|
weight=row.weight,
|
|
weight_unit=row.weight_unit,
|
|
amount=row.amount,
|
|
price=row.price
|
|
)
|
|
for row in grocery_data
|
|
]
|
|
|
|
return schemas.ShoppingEventResponse(
|
|
id=event.id,
|
|
shop_id=event.shop_id,
|
|
date=event.date,
|
|
total_amount=event.total_amount,
|
|
notes=event.notes,
|
|
created_at=event.created_at,
|
|
shop=event.shop,
|
|
groceries=groceries_with_data
|
|
)
|
|
|
|
# Root endpoint
|
|
@app.get("/")
|
|
def read_root():
|
|
return {"message": "Grocery Tracker API", "version": "1.0.0"}
|
|
|
|
# Grocery endpoints
|
|
@app.post("/groceries/", response_model=schemas.Grocery)
|
|
def create_grocery(grocery: schemas.GroceryCreate, db: Session = Depends(get_db)):
|
|
db_grocery = models.Grocery(**grocery.dict())
|
|
db.add(db_grocery)
|
|
db.commit()
|
|
db.refresh(db_grocery)
|
|
return db_grocery
|
|
|
|
@app.get("/groceries/", response_model=List[schemas.Grocery])
|
|
def read_groceries(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
groceries = db.query(models.Grocery).offset(skip).limit(limit).all()
|
|
return groceries
|
|
|
|
@app.get("/groceries/{grocery_id}", response_model=schemas.Grocery)
|
|
def read_grocery(grocery_id: int, db: Session = Depends(get_db)):
|
|
grocery = db.query(models.Grocery).filter(models.Grocery.id == grocery_id).first()
|
|
if grocery is None:
|
|
raise HTTPException(status_code=404, detail="Grocery not found")
|
|
return grocery
|
|
|
|
@app.put("/groceries/{grocery_id}", response_model=schemas.Grocery)
|
|
def update_grocery(grocery_id: int, grocery_update: schemas.GroceryUpdate, db: Session = Depends(get_db)):
|
|
grocery = db.query(models.Grocery).filter(models.Grocery.id == grocery_id).first()
|
|
if grocery is None:
|
|
raise HTTPException(status_code=404, detail="Grocery not found")
|
|
|
|
update_data = grocery_update.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(grocery, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(grocery)
|
|
return grocery
|
|
|
|
@app.delete("/groceries/{grocery_id}")
|
|
def delete_grocery(grocery_id: int, db: Session = Depends(get_db)):
|
|
grocery = db.query(models.Grocery).filter(models.Grocery.id == grocery_id).first()
|
|
if grocery is None:
|
|
raise HTTPException(status_code=404, detail="Grocery not found")
|
|
|
|
db.delete(grocery)
|
|
db.commit()
|
|
return {"message": "Grocery deleted successfully"}
|
|
|
|
# Shop endpoints
|
|
@app.post("/shops/", response_model=schemas.Shop)
|
|
def create_shop(shop: schemas.ShopCreate, db: Session = Depends(get_db)):
|
|
db_shop = models.Shop(**shop.dict())
|
|
db.add(db_shop)
|
|
db.commit()
|
|
db.refresh(db_shop)
|
|
return db_shop
|
|
|
|
@app.get("/shops/", response_model=List[schemas.Shop])
|
|
def read_shops(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
shops = db.query(models.Shop).offset(skip).limit(limit).all()
|
|
return shops
|
|
|
|
@app.get("/shops/{shop_id}", response_model=schemas.Shop)
|
|
def read_shop(shop_id: int, db: Session = Depends(get_db)):
|
|
shop = db.query(models.Shop).filter(models.Shop.id == shop_id).first()
|
|
if shop is None:
|
|
raise HTTPException(status_code=404, detail="Shop not found")
|
|
return shop
|
|
|
|
@app.put("/shops/{shop_id}", response_model=schemas.Shop)
|
|
def update_shop(shop_id: int, shop_update: schemas.ShopUpdate, db: Session = Depends(get_db)):
|
|
shop = db.query(models.Shop).filter(models.Shop.id == shop_id).first()
|
|
if shop is None:
|
|
raise HTTPException(status_code=404, detail="Shop not found")
|
|
|
|
update_data = shop_update.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(shop, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(shop)
|
|
return shop
|
|
|
|
@app.delete("/shops/{shop_id}")
|
|
def delete_shop(shop_id: int, db: Session = Depends(get_db)):
|
|
shop = db.query(models.Shop).filter(models.Shop.id == shop_id).first()
|
|
if shop is None:
|
|
raise HTTPException(status_code=404, detail="Shop not found")
|
|
|
|
db.delete(shop)
|
|
db.commit()
|
|
return {"message": "Shop deleted successfully"}
|
|
|
|
# Shopping Event endpoints
|
|
@app.post("/shopping-events/", response_model=schemas.ShoppingEventResponse)
|
|
def create_shopping_event(event: schemas.ShoppingEventCreate, db: Session = Depends(get_db)):
|
|
# Verify shop exists
|
|
shop = db.query(models.Shop).filter(models.Shop.id == event.shop_id).first()
|
|
if shop is None:
|
|
raise HTTPException(status_code=404, detail="Shop not found")
|
|
|
|
# Create shopping event
|
|
db_event = models.ShoppingEvent(
|
|
shop_id=event.shop_id,
|
|
date=event.date,
|
|
total_amount=event.total_amount,
|
|
notes=event.notes
|
|
)
|
|
db.add(db_event)
|
|
db.commit()
|
|
db.refresh(db_event)
|
|
|
|
# Add groceries to the event
|
|
for grocery_item in event.groceries:
|
|
grocery = db.query(models.Grocery).filter(models.Grocery.id == grocery_item.grocery_id).first()
|
|
if grocery is None:
|
|
raise HTTPException(status_code=404, detail=f"Grocery with id {grocery_item.grocery_id} not found")
|
|
|
|
# Insert into association table
|
|
db.execute(
|
|
models.shopping_event_groceries.insert().values(
|
|
shopping_event_id=db_event.id,
|
|
grocery_id=grocery_item.grocery_id,
|
|
amount=grocery_item.amount,
|
|
price=grocery_item.price
|
|
)
|
|
)
|
|
|
|
db.commit()
|
|
db.refresh(db_event)
|
|
return build_shopping_event_response(db_event, db)
|
|
|
|
@app.get("/shopping-events/", response_model=List[schemas.ShoppingEventResponse])
|
|
def read_shopping_events(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
events = db.query(models.ShoppingEvent).order_by(models.ShoppingEvent.created_at.desc()).offset(skip).limit(limit).all()
|
|
return [build_shopping_event_response(event, db) for event in events]
|
|
|
|
@app.get("/shopping-events/{event_id}", response_model=schemas.ShoppingEventResponse)
|
|
def read_shopping_event(event_id: int, db: Session = Depends(get_db)):
|
|
event = db.query(models.ShoppingEvent).filter(models.ShoppingEvent.id == event_id).first()
|
|
if event is None:
|
|
raise HTTPException(status_code=404, detail="Shopping event not found")
|
|
return build_shopping_event_response(event, db)
|
|
|
|
@app.put("/shopping-events/{event_id}", response_model=schemas.ShoppingEventResponse)
|
|
def update_shopping_event(event_id: int, event_update: schemas.ShoppingEventCreate, db: Session = Depends(get_db)):
|
|
# Get the existing event
|
|
event = db.query(models.ShoppingEvent).filter(models.ShoppingEvent.id == event_id).first()
|
|
if event is None:
|
|
raise HTTPException(status_code=404, detail="Shopping event not found")
|
|
|
|
# Verify shop exists
|
|
shop = db.query(models.Shop).filter(models.Shop.id == event_update.shop_id).first()
|
|
if shop is None:
|
|
raise HTTPException(status_code=404, detail="Shop not found")
|
|
|
|
# Update the shopping event
|
|
event.shop_id = event_update.shop_id
|
|
event.date = event_update.date
|
|
event.total_amount = event_update.total_amount
|
|
event.notes = event_update.notes
|
|
|
|
# Remove existing grocery associations
|
|
db.execute(
|
|
models.shopping_event_groceries.delete().where(
|
|
models.shopping_event_groceries.c.shopping_event_id == event_id
|
|
)
|
|
)
|
|
|
|
# Add new grocery associations
|
|
for grocery_item in event_update.groceries:
|
|
grocery = db.query(models.Grocery).filter(models.Grocery.id == grocery_item.grocery_id).first()
|
|
if grocery is None:
|
|
raise HTTPException(status_code=404, detail=f"Grocery with id {grocery_item.grocery_id} not found")
|
|
|
|
# Insert into association table
|
|
db.execute(
|
|
models.shopping_event_groceries.insert().values(
|
|
shopping_event_id=event_id,
|
|
grocery_id=grocery_item.grocery_id,
|
|
amount=grocery_item.amount,
|
|
price=grocery_item.price
|
|
)
|
|
)
|
|
|
|
db.commit()
|
|
db.refresh(event)
|
|
return build_shopping_event_response(event, db)
|
|
|
|
@app.delete("/shopping-events/{event_id}")
|
|
def delete_shopping_event(event_id: int, db: Session = Depends(get_db)):
|
|
event = db.query(models.ShoppingEvent).filter(models.ShoppingEvent.id == event_id).first()
|
|
if event is None:
|
|
raise HTTPException(status_code=404, detail="Shopping event not found")
|
|
|
|
# Delete grocery associations first
|
|
db.execute(
|
|
models.shopping_event_groceries.delete().where(
|
|
models.shopping_event_groceries.c.shopping_event_id == event_id
|
|
)
|
|
)
|
|
|
|
# Delete the shopping event
|
|
db.delete(event)
|
|
db.commit()
|
|
return {"message": "Shopping event deleted successfully"}
|
|
|
|
# Statistics endpoints
|
|
@app.get("/stats/categories", response_model=List[schemas.CategoryStats])
|
|
def get_category_stats(db: Session = Depends(get_db)):
|
|
# This would need more complex SQL query - placeholder for now
|
|
return []
|
|
|
|
@app.get("/stats/shops", response_model=List[schemas.ShopStats])
|
|
def get_shop_stats(db: Session = Depends(get_db)):
|
|
# This would need more complex SQL query - placeholder for now
|
|
return []
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |