from typing import Any, List from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session, joinedload from app import models, schemas from app.api import deps router = APIRouter() @router.get("/", response_model=List[schemas.CartItemWithProduct]) def read_cart_items( db: Session = Depends(deps.get_db), current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Retrieve cart items for the current user. """ cart_items = ( db.query(models.CartItem) .filter(models.CartItem.user_id == current_user.id) .options(joinedload(models.CartItem.product)) .all() ) return cart_items @router.post("/", response_model=schemas.CartItem) def add_cart_item( *, db: Session = Depends(deps.get_db), cart_item_in: schemas.CartItemCreate, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Add item to cart. """ # Check if product exists and is active product = db.query(models.Product).filter( models.Product.id == cart_item_in.product_id, models.Product.is_active == True ).first() if not product: raise HTTPException( status_code=404, detail="Product not found or inactive", ) # Check if product is in stock if product.stock < cart_item_in.quantity: raise HTTPException( status_code=400, detail=f"Not enough stock available. Only {product.stock} items left.", ) # Check if item already in cart, update quantity if it is existing_item = db.query(models.CartItem).filter( models.CartItem.user_id == current_user.id, models.CartItem.product_id == cart_item_in.product_id ).first() if existing_item: existing_item.quantity += cart_item_in.quantity db.add(existing_item) db.commit() db.refresh(existing_item) return existing_item # Create new cart item cart_item = models.CartItem( user_id=current_user.id, product_id=cart_item_in.product_id, quantity=cart_item_in.quantity, ) db.add(cart_item) db.commit() db.refresh(cart_item) return cart_item @router.put("/{cart_item_id}", response_model=schemas.CartItem) def update_cart_item( *, db: Session = Depends(deps.get_db), cart_item_id: int, cart_item_in: schemas.CartItemUpdate, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Update quantity of cart item. """ cart_item = db.query(models.CartItem).filter( models.CartItem.id == cart_item_id, models.CartItem.user_id == current_user.id ).first() if not cart_item: raise HTTPException( status_code=404, detail="Cart item not found", ) # Check if product has enough stock product = db.query(models.Product).filter(models.Product.id == cart_item.product_id).first() if not product or product.stock < cart_item_in.quantity: raise HTTPException( status_code=400, detail=f"Not enough stock available. Only {product.stock if product else 0} items left.", ) cart_item.quantity = cart_item_in.quantity db.add(cart_item) db.commit() db.refresh(cart_item) return cart_item @router.delete("/{cart_item_id}", response_model=schemas.CartItem) def delete_cart_item( *, db: Session = Depends(deps.get_db), cart_item_id: int, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Delete cart item. """ cart_item = db.query(models.CartItem).filter( models.CartItem.id == cart_item_id, models.CartItem.user_id == current_user.id ).first() if not cart_item: raise HTTPException( status_code=404, detail="Cart item not found", ) db.delete(cart_item) db.commit() return cart_item @router.delete("/", response_model=List[schemas.CartItem]) def clear_cart( db: Session = Depends(deps.get_db), current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Clear all items from cart. """ cart_items = db.query(models.CartItem).filter( models.CartItem.user_id == current_user.id ).all() for item in cart_items: db.delete(item) db.commit() return cart_items