from typing import Any, List from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query from sqlalchemy.orm import Session from app import crud, models, schemas from app.api import deps router = APIRouter() @router.get("/", response_model=List[schemas.Product]) def list_products( db: Session = Depends(deps.get_db), skip: int = Query(0, description="Skip first N items"), limit: int = Query(100, description="Limit the number of items returned"), active_only: bool = Query(False, description="Only return active products"), ) -> Any: """ Retrieve products with optional filtering """ if active_only: products = crud.product.get_active(db, skip=skip, limit=limit) else: products = crud.product.get_multi(db, skip=skip, limit=limit) return products @router.post("/", response_model=schemas.Product) def create_product( *, db: Session = Depends(deps.get_db), product_in: schemas.ProductCreate, inventory_quantity: int = Body(0, description="Initial inventory quantity"), current_user: models.User = Depends(deps.get_current_active_superuser), ) -> Any: """ Create new product with optional initial inventory """ product = crud.product.get_by_sku(db, sku=product_in.sku) if product: raise HTTPException( status_code=400, detail="A product with this SKU already exists", ) product = crud.product.create_with_inventory( db, obj_in=product_in, inventory_quantity=inventory_quantity ) return product @router.get("/{product_id}", response_model=schemas.ProductWithInventory) def get_product( *, db: Session = Depends(deps.get_db), product_id: int = Path(..., description="The ID of the product to get"), ) -> Any: """ Get a specific product by id with inventory information """ product = crud.product.get(db, id=product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") # Get inventory information inventory = crud.inventory.get_by_product_id(db, product_id=product_id) inventory_quantity = inventory.quantity if inventory else 0 # Create combined response product_data = schemas.Product.from_orm(product) product_with_inventory = schemas.ProductWithInventory( **product_data.dict(), inventory_quantity=inventory_quantity ) return product_with_inventory @router.put("/{product_id}", response_model=schemas.Product) def update_product( *, db: Session = Depends(deps.get_db), product_id: int = Path(..., description="The ID of the product to update"), product_in: schemas.ProductUpdate, current_user: models.User = Depends(deps.get_current_active_superuser), ) -> Any: """ Update a product """ product = crud.product.get(db, id=product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") # If SKU is being updated, check for uniqueness if product_in.sku and product_in.sku != product.sku: existing = crud.product.get_by_sku(db, sku=product_in.sku) if existing: raise HTTPException( status_code=400, detail="A product with this SKU already exists", ) product = crud.product.update(db, db_obj=product, obj_in=product_in) return product @router.delete("/{product_id}", response_model=schemas.Product) def delete_product( *, db: Session = Depends(deps.get_db), product_id: int = Path(..., description="The ID of the product to delete"), current_user: models.User = Depends(deps.get_current_active_superuser), ) -> Any: """ Delete a product """ product = crud.product.get(db, id=product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") # Check for any pending orders with this product # TODO: Implement order item check product = crud.product.remove(db, id=product_id) return product