Automated Action b078a91dd3 Implement simple ecommerce API with FastAPI and SQLite
- Setup project structure and FastAPI application
- Create SQLAlchemy models for users, products, carts, and orders
- Implement Alembic migrations
- Add CRUD operations and endpoints for all resources
- Setup authentication with JWT
- Add role-based access control
- Update documentation
2025-06-12 17:15:18 +00:00

148 lines
4.2 KiB
Python

from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.v1.deps import get_current_active_user
from app.core.deps import get_db
from app.crud.cart import cart_item as cart_item_crud
from app.crud.product import product as product_crud
from app.models.user import User
from app.schemas.cart import (
Cart,
CartItem,
CartItemCreate,
CartItemUpdate,
CartItemWithProduct,
)
from app.schemas.product import Product
router = APIRouter()
@router.get("/", response_model=Cart)
def read_cart(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve current user's cart.
"""
cart_items = cart_item_crud.get_user_cart(db, user_id=current_user.id)
# Fetch product details for each cart item
items_with_products = []
total = 0.0
for item in cart_items:
product = product_crud.get(db, id=item.product_id)
if product:
item_with_product = CartItemWithProduct.from_orm(item)
item_with_product.product = Product.from_orm(product)
items_with_products.append(item_with_product)
total += product.price * item.quantity
return {"items": items_with_products, "total": total}
@router.post("/items", response_model=CartItem)
def add_cart_item(
*,
db: Session = Depends(get_db),
item_in: CartItemCreate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Add item to cart.
"""
# Check if product exists and is in stock
product = product_crud.get(db, id=item_in.product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found",
)
if product.stock < item_in.quantity:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Not enough stock available. Only {product.stock} items left.",
)
# Create or update cart item
cart_item = cart_item_crud.create_or_update(
db, user_id=current_user.id, obj_in=item_in
)
return cart_item
@router.put("/items/{cart_item_id}", response_model=CartItem)
def update_cart_item(
*,
db: Session = Depends(get_db),
cart_item_id: int,
item_in: CartItemUpdate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Update quantity of cart item.
"""
cart_item = cart_item_crud.get(db, id=cart_item_id)
if not cart_item or cart_item.user_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Cart item not found",
)
# Check if product has enough stock
product = product_crud.get(db, id=cart_item.product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found",
)
if product.stock < item_in.quantity:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Not enough stock available. Only {product.stock} items left.",
)
cart_item = cart_item_crud.update(db, db_obj=cart_item, obj_in=item_in)
return cart_item
@router.delete(
"/items/{cart_item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None
)
def remove_cart_item(
*,
db: Session = Depends(get_db),
cart_item_id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Remove item from cart.
"""
cart_item = cart_item_crud.get(db, id=cart_item_id)
if not cart_item or cart_item.user_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Cart item not found",
)
cart_item_crud.remove(db, id=cart_item_id)
return None
@router.delete("/", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def clear_cart(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Clear all items from cart.
"""
cart_item_crud.clear_cart(db, user_id=current_user.id)
return None