4458f5320b
- Implemented user authentication with JWT tokens - Created product management endpoints - Added shopping cart functionality - Implemented order management system - Setup database models with SQLAlchemy - Created alembic migrations - Added health check endpoint generated with BackendIM... (backend.im)
244 lines
7.3 KiB
Python
244 lines
7.3 KiB
Python
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
|
|
from app.models.order import OrderStatus
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[schemas.Order])
|
|
def read_orders(
|
|
db: Session = Depends(deps.get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Retrieve orders for current user.
|
|
"""
|
|
if current_user.is_admin:
|
|
orders = db.query(models.Order).offset(skip).limit(limit).all()
|
|
else:
|
|
orders = (
|
|
db.query(models.Order)
|
|
.filter(models.Order.user_id == current_user.id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return orders
|
|
|
|
|
|
@router.post("/", response_model=schemas.Order)
|
|
def create_order(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
order_in: schemas.OrderCreate,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Create new order. Can create from provided items or from user's cart.
|
|
"""
|
|
# If no items provided, use cart
|
|
if not order_in.items or len(order_in.items) == 0:
|
|
cart_items = (
|
|
db.query(models.CartItem)
|
|
.filter(models.CartItem.user_id == current_user.id)
|
|
.all()
|
|
)
|
|
|
|
if not cart_items or len(cart_items) == 0:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Cart is empty and no items provided",
|
|
)
|
|
|
|
# Calculate total and create order
|
|
total_amount = 0.0
|
|
order_items = []
|
|
|
|
for cart_item in cart_items:
|
|
product = db.query(models.Product).filter(models.Product.id == cart_item.product_id).first()
|
|
|
|
if not product or not product.is_active:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Product with id {cart_item.product_id} not found or is inactive",
|
|
)
|
|
|
|
if product.stock < cart_item.quantity:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Not enough stock for product '{product.name}'. Requested: {cart_item.quantity}, Available: {product.stock}",
|
|
)
|
|
|
|
item_total = product.price * cart_item.quantity
|
|
total_amount += item_total
|
|
|
|
order_items.append(
|
|
models.OrderItem(
|
|
product_id=product.id,
|
|
quantity=cart_item.quantity,
|
|
unit_price=product.price,
|
|
)
|
|
)
|
|
|
|
# Update product stock
|
|
product.stock -= cart_item.quantity
|
|
db.add(product)
|
|
|
|
# Remove from cart
|
|
db.delete(cart_item)
|
|
else:
|
|
# Create order from provided items
|
|
total_amount = 0.0
|
|
order_items = []
|
|
|
|
for item in order_in.items:
|
|
product = db.query(models.Product).filter(
|
|
models.Product.id == item.product_id,
|
|
models.Product.is_active == True
|
|
).first()
|
|
|
|
if not product:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Product with id {item.product_id} not found or is inactive",
|
|
)
|
|
|
|
if product.stock < item.quantity:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Not enough stock for product '{product.name}'. Requested: {item.quantity}, Available: {product.stock}",
|
|
)
|
|
|
|
item_total = product.price * item.quantity
|
|
total_amount += item_total
|
|
|
|
order_items.append(
|
|
models.OrderItem(
|
|
product_id=product.id,
|
|
quantity=item.quantity,
|
|
unit_price=product.price,
|
|
)
|
|
)
|
|
|
|
# Update product stock
|
|
product.stock -= item.quantity
|
|
db.add(product)
|
|
|
|
# Create order
|
|
order = models.Order(
|
|
user_id=current_user.id,
|
|
total_amount=total_amount,
|
|
status=OrderStatus.PENDING,
|
|
shipping_address=order_in.shipping_address,
|
|
)
|
|
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
|
|
# Add order items
|
|
for item in order_items:
|
|
item.order_id = order.id
|
|
db.add(item)
|
|
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order
|
|
|
|
|
|
@router.get("/{order_id}", response_model=schemas.OrderWithItems)
|
|
def read_order(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
order_id: int,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get a specific order by id.
|
|
"""
|
|
order = (
|
|
db.query(models.Order)
|
|
.options(joinedload(models.Order.items).joinedload(models.OrderItem.product))
|
|
.filter(models.Order.id == order_id)
|
|
.first()
|
|
)
|
|
|
|
if not order:
|
|
raise HTTPException(status_code=404, detail="Order not found")
|
|
|
|
if order.user_id != current_user.id and not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
|
|
return order
|
|
|
|
|
|
@router.put("/{order_id}/status", response_model=schemas.Order)
|
|
def update_order_status(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
order_id: int,
|
|
status: OrderStatus,
|
|
current_user: models.User = Depends(deps.get_current_active_admin),
|
|
) -> Any:
|
|
"""
|
|
Update an order's status. Admin only.
|
|
"""
|
|
order = db.query(models.Order).filter(models.Order.id == order_id).first()
|
|
|
|
if not order:
|
|
raise HTTPException(status_code=404, detail="Order not found")
|
|
|
|
order.status = status
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order
|
|
|
|
|
|
@router.delete("/{order_id}", response_model=schemas.Order)
|
|
def cancel_order(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
order_id: int,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Cancel an order. Only possible if status is pending.
|
|
"""
|
|
order = db.query(models.Order).filter(
|
|
models.Order.id == order_id
|
|
).first()
|
|
|
|
if not order:
|
|
raise HTTPException(status_code=404, detail="Order not found")
|
|
|
|
if order.user_id != current_user.id and not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
|
|
if order.status != OrderStatus.PENDING:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Cannot cancel order with status '{order.status}'. Only pending orders can be cancelled."
|
|
)
|
|
|
|
# Return items to stock
|
|
order_items = db.query(models.OrderItem).filter(models.OrderItem.order_id == order.id).all()
|
|
|
|
for item in order_items:
|
|
product = db.query(models.Product).filter(models.Product.id == item.product_id).first()
|
|
if product:
|
|
product.stock += item.quantity
|
|
db.add(product)
|
|
|
|
order.status = OrderStatus.CANCELLED
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order |