from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from app.db.session import get_db from app.models.product import Product, Category from app.schemas.product import ( ProductCreate, ProductUpdate, ProductInDB, CategoryCreate, CategoryUpdate, CategoryInDB ) from app.api.deps import get_product, get_category router = APIRouter(prefix="/api/v1") # Category endpoints @router.post("/categories/", response_model=CategoryInDB, status_code=status.HTTP_201_CREATED) def create_category( *, db: Session = Depends(get_db), category_in: CategoryCreate ): """Create a new category""" try: db_category = Category(**category_in.model_dump()) db.add(db_category) db.commit() db.refresh(db_category) return db_category except IntegrityError: db.rollback() raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Category with this name already exists" ) @router.get("/categories/", response_model=List[CategoryInDB]) def read_categories( *, db: Session = Depends(get_db), skip: int = 0, limit: int = 100, search: Optional[str] = None ): """Get all categories with optional search filter""" query = db.query(Category) if search: query = query.filter(Category.name.ilike(f"%{search}%")) return query.offset(skip).limit(limit).all() @router.get("/categories/{category_id}", response_model=CategoryInDB) def read_category( *, db: Session = Depends(get_db), category: Category = Depends(get_category) ): """Get a specific category by ID""" return category @router.put("/categories/{category_id}", response_model=CategoryInDB) def update_category( *, db: Session = Depends(get_db), category: Category = Depends(get_category), category_in: CategoryUpdate ): """Update a category""" update_data = category_in.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(category, field, value) try: db.add(category) db.commit() db.refresh(category) return category except IntegrityError: db.rollback() raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Category with this name already exists" ) @router.delete("/categories/{category_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_category( *, db: Session = Depends(get_db), category: Category = Depends(get_category) ): """Delete a category""" # Check if category has products if category.products: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete category with associated products" ) db.delete(category) db.commit() return None # Product endpoints @router.post("/products/", response_model=ProductInDB, status_code=status.HTTP_201_CREATED) def create_product( *, db: Session = Depends(get_db), product_in: ProductCreate ): """Create a new product""" # Check if category exists if product_in.category_id: category = db.query(Category).filter(Category.id == product_in.category_id).first() if not category: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Category with ID {product_in.category_id} not found" ) try: db_product = Product(**product_in.model_dump()) db.add(db_product) db.commit() db.refresh(db_product) return db_product except IntegrityError: db.rollback() raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Product with this SKU or barcode already exists" ) @router.get("/products/", response_model=List[ProductInDB]) def read_products( *, db: Session = Depends(get_db), skip: int = 0, limit: int = 100, search: Optional[str] = None, category_id: Optional[int] = None ): """Get all products with optional filters""" query = db.query(Product) if search: query = query.filter( (Product.name.ilike(f"%{search}%")) | (Product.sku.ilike(f"%{search}%")) | (Product.description.ilike(f"%{search}%")) ) if category_id: query = query.filter(Product.category_id == category_id) return query.offset(skip).limit(limit).all() @router.get("/products/{product_id}", response_model=ProductInDB) def read_product( *, db: Session = Depends(get_db), product: Product = Depends(get_product) ): """Get a specific product by ID""" return product @router.put("/products/{product_id}", response_model=ProductInDB) def update_product( *, db: Session = Depends(get_db), product: Product = Depends(get_product), product_in: ProductUpdate ): """Update a product""" # Check if category exists if product_in.category_id is not None: category = db.query(Category).filter(Category.id == product_in.category_id).first() if not category and product_in.category_id is not None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Category with ID {product_in.category_id} not found" ) update_data = product_in.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(product, field, value) try: db.add(product) db.commit() db.refresh(product) return product except IntegrityError: db.rollback() raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Product with this SKU or barcode already exists" ) @router.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_product( *, db: Session = Depends(get_db), product: Product = Depends(get_product) ): """Delete a product""" # Check if product has inventory items if product.inventory_items: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete product with existing inventory" ) # Check if product is used in purchase items if product.purchase_items: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete product with purchase history" ) db.delete(product) db.commit() return None