d2b80dacc4
- Set up FastAPI application structure - Create database models for User, Product, Cart, CartItem, Order, and OrderItem - Set up Alembic for database migrations - Create Pydantic schemas for request/response models - Implement API endpoints for products, cart operations, and checkout process - Add health endpoint - Update README with project details and documentation
249 lines
7.6 KiB
Python
249 lines
7.6 KiB
Python
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Header, Cookie, status, Response
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import get_db
|
|
from app.models.cart import CartItem
|
|
from app.schemas.cart import (
|
|
Cart as CartSchema,
|
|
CartItemCreate,
|
|
CartItemUpdate,
|
|
CartItemWithProduct,
|
|
AddToCartResponse
|
|
)
|
|
from app.crud import product, cart, cart_item
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def get_cart_id(
|
|
db: Session = Depends(get_db),
|
|
user_id: Optional[int] = None,
|
|
session_id: Optional[str] = Cookie(None),
|
|
x_session_id: Optional[str] = Header(None)
|
|
) -> int:
|
|
"""
|
|
Get or create an active cart for the current user/session.
|
|
|
|
This function will:
|
|
1. Try to get the active cart for the authenticated user (if logged in)
|
|
2. If no user, try to get the active cart for the current session
|
|
3. If no active cart exists, create a new one
|
|
|
|
Returns the cart ID.
|
|
"""
|
|
# If the session_id is not in cookies, try to get it from headers
|
|
current_session_id = session_id or x_session_id
|
|
|
|
# Get or create an active cart
|
|
active_cart = cart.get_or_create_active_cart(
|
|
db=db,
|
|
user_id=user_id,
|
|
session_id=current_session_id
|
|
)
|
|
|
|
return active_cart.id
|
|
|
|
|
|
@router.get("", response_model=CartSchema)
|
|
def get_current_cart(
|
|
cart_id: int = Depends(get_cart_id),
|
|
db: Session = Depends(get_db),
|
|
) -> Any:
|
|
"""
|
|
Get the current active cart with all items.
|
|
"""
|
|
current_cart = cart.get_cart_with_items(db=db, cart_id=cart_id)
|
|
if not current_cart:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Cart not found"
|
|
)
|
|
return current_cart
|
|
|
|
|
|
@router.post("/items", response_model=AddToCartResponse)
|
|
def add_item_to_cart(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
item_in: CartItemCreate,
|
|
cart_id: int = Depends(get_cart_id),
|
|
response: Response
|
|
) -> Any:
|
|
"""
|
|
Add an item to the cart.
|
|
|
|
If the item already exists in the cart, the quantity will be updated.
|
|
"""
|
|
# Check if product exists and is active
|
|
item_product = product.get(db=db, id=item_in.product_id)
|
|
if not item_product:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Product with ID {item_in.product_id} not found"
|
|
)
|
|
|
|
if not item_product.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Product with ID {item_in.product_id} is not available"
|
|
)
|
|
|
|
# Check if product is in stock
|
|
if item_product.stock_quantity < item_in.quantity:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Not enough stock available. Only {item_product.stock_quantity} items left."
|
|
)
|
|
|
|
# Get the current cart or create a new one
|
|
current_cart = cart.get(db=db, id=cart_id)
|
|
|
|
# Check if the product is already in the cart
|
|
existing_item = cart_item.get_by_cart_and_product(
|
|
db=db, cart_id=current_cart.id, product_id=item_in.product_id
|
|
)
|
|
|
|
if existing_item:
|
|
# Update the quantity
|
|
new_quantity = existing_item.quantity + item_in.quantity
|
|
|
|
# Check stock for the new total quantity
|
|
if item_product.stock_quantity < new_quantity:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Not enough stock available. Only {item_product.stock_quantity} items left."
|
|
)
|
|
|
|
item_update = CartItemUpdate(quantity=new_quantity)
|
|
updated_item = cart_item.update(db=db, db_obj=existing_item, obj_in=item_update)
|
|
|
|
# Fetch the updated item with product details
|
|
result = db.query(CartItem).filter(CartItem.id == updated_item.id).first()
|
|
result.product = item_product
|
|
|
|
# Set a cookie with the session ID if applicable
|
|
if current_cart.session_id:
|
|
response.set_cookie(key="session_id", value=current_cart.session_id, httponly=True)
|
|
|
|
return AddToCartResponse(
|
|
cart_id=current_cart.id,
|
|
message="Item quantity updated in cart",
|
|
cart_item=result
|
|
)
|
|
else:
|
|
# Add the new item to the cart
|
|
new_item = cart_item.create_with_cart(
|
|
db=db,
|
|
obj_in=item_in,
|
|
cart_id=current_cart.id,
|
|
product_price=float(item_product.price)
|
|
)
|
|
|
|
# Fetch the new item with product details
|
|
result = db.query(CartItem).filter(CartItem.id == new_item.id).first()
|
|
result.product = item_product
|
|
|
|
# Set a cookie with the session ID if applicable
|
|
if current_cart.session_id:
|
|
response.set_cookie(key="session_id", value=current_cart.session_id, httponly=True)
|
|
|
|
return AddToCartResponse(
|
|
cart_id=current_cart.id,
|
|
message="Item added to cart",
|
|
cart_item=result
|
|
)
|
|
|
|
|
|
@router.put("/items/{item_id}", response_model=CartItemWithProduct)
|
|
def update_cart_item(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
item_id: int,
|
|
item_in: CartItemUpdate,
|
|
cart_id: int = Depends(get_cart_id)
|
|
) -> Any:
|
|
"""
|
|
Update the quantity of an item in the cart.
|
|
"""
|
|
# Check if the item exists in the current cart
|
|
item = db.query(CartItem).filter(
|
|
CartItem.id == item_id,
|
|
CartItem.cart_id == cart_id
|
|
).first()
|
|
|
|
if not item:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Item with ID {item_id} not found in cart"
|
|
)
|
|
|
|
# Check product availability and stock
|
|
item_product = product.get(db=db, id=item.product_id)
|
|
if not item_product or not item_product.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Product is not available"
|
|
)
|
|
|
|
if item_product.stock_quantity < item_in.quantity:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Not enough stock available. Only {item_product.stock_quantity} items left."
|
|
)
|
|
|
|
# Update the item
|
|
updated_item = cart_item.update(db=db, db_obj=item, obj_in=item_in)
|
|
|
|
# Fetch the updated item with product details
|
|
result = db.query(CartItem).filter(CartItem.id == updated_item.id).first()
|
|
result.product = item_product
|
|
|
|
return result
|
|
|
|
|
|
@router.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
|
|
def remove_cart_item(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
item_id: int,
|
|
cart_id: int = Depends(get_cart_id)
|
|
) -> None:
|
|
"""
|
|
Remove an item from the cart.
|
|
"""
|
|
# Check if the item exists in the current cart
|
|
item = db.query(CartItem).filter(
|
|
CartItem.id == item_id,
|
|
CartItem.cart_id == cart_id
|
|
).first()
|
|
|
|
if not item:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Item with ID {item_id} not found in cart"
|
|
)
|
|
|
|
# Remove the item
|
|
cart_item.remove(db=db, id=item_id)
|
|
return None
|
|
|
|
|
|
@router.delete("", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
|
|
def clear_cart(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
cart_id: int = Depends(get_cart_id)
|
|
) -> None:
|
|
"""
|
|
Remove all items from the cart.
|
|
"""
|
|
# Get all items in the cart
|
|
items = cart_item.get_by_cart(db=db, cart_id=cart_id)
|
|
|
|
# Remove all items
|
|
for item in items:
|
|
cart_item.remove(db=db, id=item.id)
|
|
|
|
return None |