from typing import Any, List from fastapi import APIRouter, 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.Customer]) def list_customers( 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"), current_user: models.User = Depends(deps.get_current_active_superuser), ) -> Any: """ Retrieve customers """ customers = crud.customer.get_multi(db, skip=skip, limit=limit) return customers @router.post("/", response_model=schemas.Customer) def create_customer( *, db: Session = Depends(deps.get_db), customer_in: schemas.CustomerCreate, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Create new customer """ # Check if email already exists customer = crud.customer.get_by_email(db, email=customer_in.email) if customer: raise HTTPException( status_code=400, detail="A customer with this email already exists", ) # For now, we'll create the customer without a Stripe ID # In a real app, we'd create the Stripe customer first customer = crud.customer.create(db, obj_in=customer_in) return customer @router.get("/{customer_id}", response_model=schemas.Customer) def get_customer( *, db: Session = Depends(deps.get_db), customer_id: int = Path(..., description="The ID of the customer to get"), current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Get a specific customer by id """ customer = crud.customer.get(db, id=customer_id) if not customer: raise HTTPException(status_code=404, detail="Customer not found") # Check if user is superuser or if they're accessing their own customer record if not crud.user.is_superuser(current_user): if not customer.user_id or customer.user_id != current_user.id: raise HTTPException( status_code=403, detail="Not enough permissions to access this customer" ) return customer @router.put("/{customer_id}", response_model=schemas.Customer) def update_customer( *, db: Session = Depends(deps.get_db), customer_id: int = Path(..., description="The ID of the customer to update"), customer_in: schemas.CustomerUpdate, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Update a customer """ customer = crud.customer.get(db, id=customer_id) if not customer: raise HTTPException(status_code=404, detail="Customer not found") # Check if user is superuser or if they're updating their own customer record if not crud.user.is_superuser(current_user): if not customer.user_id or customer.user_id != current_user.id: raise HTTPException( status_code=403, detail="Not enough permissions to update this customer" ) # If email is being updated, check if it's already in use if customer_in.email and customer_in.email != customer.email: existing = crud.customer.get_by_email(db, email=customer_in.email) if existing and existing.id != customer_id: raise HTTPException( status_code=400, detail="A customer with this email already exists", ) customer = crud.customer.update(db, db_obj=customer, obj_in=customer_in) return customer @router.delete("/{customer_id}", response_model=schemas.Customer) def delete_customer( *, db: Session = Depends(deps.get_db), customer_id: int = Path(..., description="The ID of the customer to delete"), current_user: models.User = Depends(deps.get_current_active_superuser), ) -> Any: """ Delete a customer (admin only) """ customer = crud.customer.get(db, id=customer_id) if not customer: raise HTTPException(status_code=404, detail="Customer not found") # In a real app, we'd need to handle Stripe customer deletion and check for orders customer = crud.customer.remove(db, id=customer_id) return customer