Spaces:
Paused
Paused
File size: 6,636 Bytes
3e5c7dd a3b84cc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
from fastapi import APIRouter, HTTPException, status, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List, Optional
from ..core.dependencies import get_current_active_user
from ..db.database import get_db
from ..db.models import Order, Product, OrderItem, User
from ..db.schemas import OrderCreate, OrderInDB
from datetime import datetime
router = APIRouter()
@router.post("/", response_model=OrderInDB)
async def create_order(
order: OrderCreate,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> OrderInDB:
# Ensure user belongs to the branch they're creating the order for
if current_user.branch_id != order.branch_id and not current_user.is_superuser:
raise HTTPException(
status_code=403,
detail="You can only create orders for your own branch"
)
# Calculate total and validate products
total = 0
order_items = []
for item in order.items:
# Get product
stmt = select(Product).where(
Product.id == item.product_id,
Product.branch_id == order.branch_id # Ensure product belongs to the same branch
)
result = await db.execute(stmt)
product = result.scalar_one_or_none()
if not product:
raise HTTPException(
status_code=404,
detail=f"Product {item.product_id} not found in this branch"
)
if product.inventory_count < item.quantity:
raise HTTPException(
status_code=400,
detail=f"Insufficient inventory for product {item.product_id}"
)
# Update inventory
product.inventory_count -= item.quantity
total += product.price * item.quantity
# Create order item
order_item = OrderItem(
product_id=item.product_id,
quantity=item.quantity,
price=product.price
)
order_items.append(order_item)
# Create order
db_order = Order(
customer_id=order.customer_id,
branch_id=order.branch_id,
total_amount=total,
status="pending",
items=order_items
)
db.add(db_order)
await db.commit()
await db.refresh(db_order)
return db_order
@router.get("/", response_model=List[OrderInDB])
async def list_orders(
skip: int = 0,
limit: int = 10,
status: Optional[str] = None,
branch_id: Optional[int] = Query(None, description="Filter orders by branch"),
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> List[OrderInDB]:
query = select(Order)
# Filter by status if provided
if status:
query = query.where(Order.status == status)
# Filter by branch if provided, otherwise use user's branch
if branch_id:
if not current_user.is_superuser and branch_id != current_user.branch_id:
raise HTTPException(
status_code=403,
detail="You can only view orders from your own branch"
)
query = query.where(Order.branch_id == branch_id)
elif not current_user.is_superuser:
# Non-superusers can only see orders from their branch
query = query.where(Order.branch_id == current_user.branch_id)
query = query.offset(skip).limit(limit)
result = await db.execute(query)
return result.scalars().all()
@router.get("/{order_id}", response_model=OrderInDB)
async def get_order(
order_id: int,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> OrderInDB:
stmt = select(Order).where(Order.id == order_id)
result = await db.execute(stmt)
order = result.scalar_one_or_none()
if not order:
raise HTTPException(status_code=404, detail="Order not found")
# Check if user has access to this order's branch
if not current_user.is_superuser and order.branch_id != current_user.branch_id:
raise HTTPException(status_code=403, detail="You cannot access orders from other branches")
return order
@router.put("/{order_id}/status", response_model=OrderInDB)
async def update_order_status(
order_id: int,
status: str,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> OrderInDB:
valid_statuses = ["pending", "processing", "shipped", "delivered", "cancelled"]
if status not in valid_statuses:
raise HTTPException(status_code=400, detail="Invalid status")
stmt = select(Order).where(Order.id == order_id)
result = await db.execute(stmt)
order = result.scalar_one_or_none()
if not order:
raise HTTPException(status_code=404, detail="Order not found")
# Check if user has access to this order's branch
if not current_user.is_superuser and order.branch_id != current_user.branch_id:
raise HTTPException(status_code=403, detail="You cannot modify orders from other branches")
order.status = status
order.updated_at = datetime.utcnow()
await db.commit()
await db.refresh(order)
return order
@router.delete("/{order_id}")
async def delete_order(
order_id: int,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
# Get the order
stmt = select(Order).where(Order.id == order_id)
result = await db.execute(stmt)
order = result.scalar_one_or_none()
if not order:
raise HTTPException(status_code=404, detail="Order not found")
# Check if user has access to this order's branch
if not current_user.is_superuser and order.branch_id != current_user.branch_id:
raise HTTPException(status_code=403, detail="You cannot delete orders from other branches")
# Restore inventory for each product
for item in order.items:
product_stmt = select(Product).where(Product.id == item.product_id)
product_result = await db.execute(product_stmt)
product = product_result.scalar_one_or_none()
if product:
product.inventory_count += item.quantity
await db.delete(order)
await db.commit()
return {"status": "success", "message": "Order deleted and inventory restored"} |