Spaces:
Paused
Paused
File size: 5,069 Bytes
8a44fbc b70ff07 | 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 | 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 Product, User
from ..db.schemas import ProductCreate, ProductInDB
router = APIRouter()
@router.post("/", response_model=ProductInDB)
async def create_product(
product: ProductCreate,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> ProductInDB:
# Ensure user belongs to the branch they're creating the product for
if current_user.branch_id != product.branch_id and not current_user.is_superuser:
raise HTTPException(
status_code=403,
detail="You can only create products for your own branch"
)
# Create product with current user as seller
product_data = product.dict()
product_data["seller_id"] = current_user.id
db_product = Product(**product_data)
db.add(db_product)
await db.commit()
await db.refresh(db_product)
return db_product
@router.get("/", response_model=List[ProductInDB])
async def list_products(
skip: int = 0,
limit: int = 10,
category: Optional[str] = None,
branch_id: Optional[int] = Query(None, description="Filter products by branch"),
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> List[ProductInDB]:
query = select(Product)
# Filter by category if provided
if category:
query = query.where(Product.category == category)
# 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 products from your own branch"
)
query = query.where(Product.branch_id == branch_id)
elif not current_user.is_superuser:
# Non-superusers can only see products from their branch
query = query.where(Product.branch_id == current_user.branch_id)
query = query.offset(skip).limit(limit)
result = await db.execute(query)
return result.scalars().all()
@router.get("/{product_id}", response_model=ProductInDB)
async def get_product(
product_id: int,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> ProductInDB:
stmt = select(Product).where(Product.id == product_id)
result = await db.execute(stmt)
product = result.scalar_one_or_none()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Check if user has access to this product's branch
if not current_user.is_superuser and product.branch_id != current_user.branch_id:
raise HTTPException(status_code=403, detail="You cannot access products from other branches")
return product
@router.put("/{product_id}", response_model=ProductInDB)
async def update_product(
product_id: int,
product_update: ProductCreate,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
) -> ProductInDB:
stmt = select(Product).where(Product.id == product_id)
result = await db.execute(stmt)
product = result.scalar_one_or_none()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Check if user has access to this product's branch
if not current_user.is_superuser and product.branch_id != current_user.branch_id:
raise HTTPException(status_code=403, detail="You cannot modify products from other branches")
# Ensure the branch isn't being changed to a different branch
if product_update.branch_id != product.branch_id:
raise HTTPException(status_code=400, detail="Cannot change product's branch")
# Update product fields
update_data = product_update.dict(exclude_unset=True)
for field, value in update_data.items():
setattr(product, field, value)
await db.commit()
await db.refresh(product)
return product
@router.delete("/{product_id}")
async def delete_product(
product_id: int,
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db)
):
stmt = select(Product).where(Product.id == product_id)
result = await db.execute(stmt)
product = result.scalar_one_or_none()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Check if user has access to this product's branch
if not current_user.is_superuser and product.branch_id != current_user.branch_id:
raise HTTPException(status_code=403, detail="You cannot delete products from other branches")
await db.delete(product)
await db.commit()
return {"status": "success", "message": "Product deleted"} |