Upload app/api/cart.py

app/api/cart.py

@@ -0,0 +1,236 @@
+from typing import Any, Optional
+from fastapi import APIRouter, Depends, HTTPException, status, Header
+from sqlalchemy.orm import Session
+from datetime import datetime, timezone
+
+from app.db.base import get_db
+from app.models.user import User
+from app.models.product import Product
+from app.models.cart import Cart, CartItem, Coupon
+from app.schemas.cart import (
+    CartResponse, CartItemCreate, CartItemUpdate, 
+    CartItemResponse, CouponApply, CouponResponse
+)
+from app.core.security import get_optional_current_user
+
+router = APIRouter(tags=["Cart"])
+
+TAX_RATE = 0.15  # 15% VAT
+SHIPPING_FLAT_RATE = 20.0  # Flat shipping rate
+FREE_SHIPPING_THRESHOLD = 500.0  # Free shipping over 500
+
+
+def calculate_cart_totals(cart: Cart) -> dict:
+    subtotal = sum(item.quantity * item.unit_price for item in cart.items)
+    
+    # Calculate discount
+    discount_amount = 0.0
+    applied_coupon = None
+    if cart.coupon and cart.coupon.is_active:
+        if cart.coupon.expires_at is None or cart.coupon.expires_at > datetime.now(timezone.utc):
+            discount_amount = subtotal * (cart.coupon.discount_percent / 100.0)
+            applied_coupon = CouponResponse.model_validate(cart.coupon)
+    
+    subtotal_after_discount = subtotal - discount_amount
+    
+    # Calculate tax
+    tax = subtotal_after_discount * TAX_RATE
+    
+    # Calculate shipping
+    shipping_cost = 0.0 if subtotal_after_discount >= FREE_SHIPPING_THRESHOLD or subtotal_after_discount == 0 else SHIPPING_FLAT_RATE
+    
+    total = subtotal_after_discount + tax + shipping_cost
+
+    return {
+        "subtotal": round(subtotal, 2),
+        "tax": round(tax, 2),
+        "shipping_cost": round(shipping_cost, 2),
+        "discount_amount": round(discount_amount, 2),
+        "total": round(total, 2),
+        "applied_coupon": applied_coupon
+    }
+
+def get_or_create_cart(db: Session, user_id: Optional[int] = None, session_id: Optional[str] = None) -> Cart:
+    """Get or create a cart for either an authenticated user or a guest session."""
+    if user_id:
+        cart = db.query(Cart).filter(Cart.user_id == user_id).first()
+        if not cart:
+            cart = Cart(user_id=user_id)
+            db.add(cart)
+            db.commit()
+            db.refresh(cart)
+        return cart
+    elif session_id:
+        cart = db.query(Cart).filter(Cart.session_id == session_id).first()
+        if not cart:
+            cart = Cart(session_id=session_id)
+            db.add(cart)
+            db.commit()
+            db.refresh(cart)
+        return cart
+    else:
+        raise HTTPException(status_code=400, detail="Either login or provide X-Cart-ID header")
+
+@router.get("/cart", response_model=dict)
+def get_cart(
+    current_user_id: Optional[int] = Depends(get_optional_current_user),
+    x_cart_id: Optional[str] = Header(None),
+    db: Session = Depends(get_db)
+) -> Any:
+    """Get the current user's (or guest session's) shopping cart."""
+    cart = get_or_create_cart(db, user_id=current_user_id, session_id=x_cart_id)
+    totals = calculate_cart_totals(cart)
+    
+    items_response = []
+    for item in cart.items:
+        item_data = CartItemResponse(
+            id=item.id,
+            cart_id=item.cart_id,
+            product_id=item.product_id,
+            quantity=item.quantity,
+            unit_price=item.unit_price,
+            total_price=item.quantity * item.unit_price,
+            product=item.product
+        )
+        items_response.append(item_data)
+        
+    return {
+        "isSuccess": True,
+        "value": CartResponse(
+            id=cart.id,
+            user_id=cart.user_id,
+            items=items_response,
+            **totals
+        ),
+        "statusCode": 200
+    }
+
+@router.post("/cart/items", response_model=dict)
+def add_item_to_cart(
+    item_in: CartItemCreate,
+    current_user_id: Optional[int] = Depends(get_optional_current_user),
+    x_cart_id: Optional[str] = Header(None),
+    db: Session = Depends(get_db)
+) -> Any:
+    """Add a product to the cart."""
+    cart = get_or_create_cart(db, user_id=current_user_id, session_id=x_cart_id)
+    
+    # Check product availability
+    product = db.query(Product).filter(Product.id == item_in.product_id).first()
+    if not product:
+        raise HTTPException(status_code=404, detail="Product not found")
+    if not product.is_active:
+        raise HTTPException(status_code=400, detail="Product is not currently available")
+        
+    # Check if item already in cart
+    cart_item = db.query(CartItem).filter(
+        CartItem.cart_id == cart.id,
+        CartItem.product_id == item_in.product_id
+    ).first()
+    
+    if cart_item:
+        if cart_item.quantity + item_in.quantity > product.stock:
+            raise HTTPException(status_code=400, detail=f"Not enough stock. Only {product.stock} available.")
+        cart_item.quantity += item_in.quantity
+        # refresh price just to be safe
+        cart_item.unit_price = product.price
+    else:
+        if item_in.quantity > product.stock:
+            raise HTTPException(status_code=400, detail=f"Not enough stock. Only {product.stock} available.")
+        cart_item = CartItem(
+            cart_id=cart.id,
+            product_id=item_in.product_id,
+            quantity=item_in.quantity,
+            unit_price=product.price
+        )
+        db.add(cart_item)
+        
+    db.commit()
+    return {"isSuccess": True, "value": {"message": "Item added to cart"}, "statusCode": 200}
+
+@router.put("/cart/items/{item_id}", response_model=dict)
+def update_cart_item(
+    item_id: int,
+    item_in: CartItemUpdate,
+    current_user_id: Optional[int] = Depends(get_optional_current_user),
+    x_cart_id: Optional[str] = Header(None),
+    db: Session = Depends(get_db)
+) -> Any:
+    """Update quantity of an item in the cart."""
+    cart = get_or_create_cart(db, user_id=current_user_id, session_id=x_cart_id)
+    
+    cart_item = db.query(CartItem).filter(
+        CartItem.id == item_id,
+        CartItem.cart_id == cart.id
+    ).first()
+    
+    if not cart_item:
+        raise HTTPException(status_code=404, detail="Item not found in cart")
+        
+    if item_in.quantity > cart_item.product.stock:
+        raise HTTPException(status_code=400, detail=f"Not enough stock. Only {cart_item.product.stock} available.")
+        
+    cart_item.quantity = item_in.quantity
+    db.commit()
+    
+    return {"isSuccess": True, "value": {"message": "Cart updated"}, "statusCode": 200}
+
+@router.delete("/cart/items/{item_id}", response_model=dict)
+def remove_cart_item(
+    item_id: int,
+    current_user_id: Optional[int] = Depends(get_optional_current_user),
+    x_cart_id: Optional[str] = Header(None),
+    db: Session = Depends(get_db)
+) -> Any:
+    """Remove an item from the cart."""
+    cart = get_or_create_cart(db, user_id=current_user_id, session_id=x_cart_id)
+    
+    cart_item = db.query(CartItem).filter(
+        CartItem.id == item_id,
+        CartItem.cart_id == cart.id
+    ).first()
+    
+    if not cart_item:
+        raise HTTPException(status_code=404, detail="Item not found in cart")
+        
+    db.delete(cart_item)
+    db.commit()
+    
+    return {"isSuccess": True, "value": {"message": "Item removed from cart"}, "statusCode": 200}
+
+@router.post("/cart/coupon", response_model=dict)
+def apply_coupon(
+    coupon_in: CouponApply,
+    current_user_id: Optional[int] = Depends(get_optional_current_user),
+    x_cart_id: Optional[str] = Header(None),
+    db: Session = Depends(get_db)
+) -> Any:
+    """Apply a discount coupon to the cart."""
+    cart = get_or_create_cart(db, user_id=current_user_id, session_id=x_cart_id)
+    
+    coupon = db.query(Coupon).filter(Coupon.code == coupon_in.code.upper()).first()
+    if not coupon or not coupon.is_active:
+        raise HTTPException(status_code=400, detail="Invalid coupon code")
+        
+    if coupon.expires_at and coupon.expires_at < datetime.now(timezone.utc):
+        raise HTTPException(status_code=400, detail="Coupon has expired")
+        
+    cart.coupon_id = coupon.id
+    db.commit()
+    
+    return {"isSuccess": True, "value": {"message": f"Coupon {coupon.code} applied successfully"}, "statusCode": 200}
+
+@router.delete("/cart/clear", response_model=dict)
+def clear_cart(
+    current_user_id: Optional[int] = Depends(get_optional_current_user),
+    x_cart_id: Optional[str] = Header(None),
+    db: Session = Depends(get_db)
+) -> Any:
+    """Remove all items from the cart and detach coupons."""
+    cart = get_or_create_cart(db, user_id=current_user_id, session_id=x_cart_id)
+    
+    db.query(CartItem).filter(CartItem.cart_id == cart.id).delete()
+    cart.coupon_id = None
+    db.commit()
+    
+    return {"isSuccess": True, "value": {"message": "Cart cleared"}, "statusCode": 200}