from typing import Any, List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app import crud, models from app.api.deps import get_current_active_user, get_db from app.schemas.customer import Customer, CustomerCreate, CustomerUpdate router = APIRouter() @router.get("/", response_model=List[Customer]) def read_customers( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, current_user: models.User = Depends(get_current_active_user), ) -> Any: """ Retrieve customers. """ customers = crud.customer.get_multi(db, skip=skip, limit=limit) return customers @router.post("/", response_model=Customer, status_code=status.HTTP_201_CREATED) def create_customer( *, db: Session = Depends(get_db), customer_in: CustomerCreate, current_user: models.User = Depends(get_current_active_user), ) -> Any: """ Create new customer. """ customer = crud.customer.get_by_email(db, email=customer_in.email) if customer: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A customer with this email already exists in the system.", ) customer = crud.customer.create(db, obj_in=customer_in) return customer @router.get("/{customer_id}", response_model=Customer) def read_customer( *, db: Session = Depends(get_db), customer_id: int, current_user: models.User = Depends(get_current_active_user), ) -> Any: """ Get customer by ID. """ customer = crud.customer.get(db, id=customer_id) if not customer: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Customer not found", ) return customer @router.put("/{customer_id}", response_model=Customer) def update_customer( *, db: Session = Depends(get_db), customer_id: int, customer_in: CustomerUpdate, current_user: models.User = Depends(get_current_active_user), ) -> Any: """ Update a customer. """ customer = crud.customer.get(db, id=customer_id) if not customer: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Customer not found", ) # Check if email exists and it's not the customer's own email if customer_in.email and customer_in.email != customer.email: db_customer = crud.customer.get_by_email(db, email=customer_in.email) if db_customer: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A customer with this email already exists in the system.", ) customer = crud.customer.update(db, db_obj=customer, obj_in=customer_in) return customer @router.delete("/{customer_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None) def delete_customer( *, db: Session = Depends(get_db), customer_id: int, current_user: models.User = Depends(get_current_active_user), ) -> Any: """ Delete a customer. """ customer = crud.customer.get(db, id=customer_id) if not customer: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Customer not found", ) crud.customer.remove(db, id=customer_id) return None