from typing import Any, List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.api.deps import get_current_active_user, get_current_admin_user, get_db from app.crud.order import order, order_item from app.models.order import OrderStatus from app.models.user import User from app.schemas.order import ( Order, OrderCreate, OrderUpdate, OrderItem, OrderItemCreate, OrderItemUpdate ) router = APIRouter() @router.get("/", response_model=List[Order]) def read_orders( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, current_user: User = Depends(get_current_active_user), ) -> Any: """ Retrieve orders. """ if current_user.is_admin: return order.get_multi(db, skip=skip, limit=limit) return order.get_by_user(db, user_id=current_user.id, skip=skip, limit=limit) @router.post("/", response_model=Order) def create_order( *, db: Session = Depends(get_db), order_in: OrderCreate, current_user: User = Depends(get_current_active_user), ) -> Any: """ Create new order. """ # Ensure user can only create orders for themselves if order_in.user_id != current_user.id and not current_user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You can only create orders for yourself", ) try: return order.create(db, obj_in=order_in) except ValueError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e), ) @router.get("/{order_id}", response_model=Order) def read_order( *, db: Session = Depends(get_db), order_id: int, current_user: User = Depends(get_current_active_user), ) -> Any: """ Get order by ID. """ db_order = order.get(db, id=order_id) if not db_order: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Order not found", ) # Ensure users can only view their own orders unless they're admins if db_order.user_id != current_user.id and not current_user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You can only view your own orders", ) return db_order @router.put("/{order_id}/status", response_model=Order) def update_order_status( *, db: Session = Depends(get_db), order_id: int, status: OrderStatus, current_user: User = Depends(get_current_admin_user), ) -> Any: """ Update order status. Only for admins. """ db_order = order.get(db, id=order_id) if not db_order: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Order not found", ) return order.update_status(db, db_obj=db_order, status=status) @router.put("/{order_id}", response_model=Order) def update_order( *, db: Session = Depends(get_db), order_id: int, order_in: OrderUpdate, current_user: User = Depends(get_current_active_user), ) -> Any: """ Update an order. """ db_order = order.get(db, id=order_id) if not db_order: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Order not found", ) # Users can only update their own orders if db_order.user_id != current_user.id and not current_user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You can only update your own orders", ) # Regular users can only update the shipping address if the order is still pending if not current_user.is_admin and db_order.status != OrderStatus.PENDING: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="You can only update pending orders", ) # Regular users can't update status if order_in.status and not current_user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You cannot update order status", ) return order.update(db, db_obj=db_order, obj_in=order_in) @router.get("/status/{status}", response_model=List[Order]) def read_orders_by_status( *, db: Session = Depends(get_db), status: OrderStatus, skip: int = 0, limit: int = 100, current_user: User = Depends(get_current_admin_user), ) -> Any: """ Get orders by status. Only for admins. """ return order.get_by_status(db, status=status, skip=skip, limit=limit)