from typing import Any, List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.core.deps import get_current_active_user from app.crud.user import get_user, get_users, update_user, delete_user from app.db.database import get_db from app.models.user import User from app.schemas.user import User as UserSchema, UserUpdate router = APIRouter( prefix="/users", tags=["users"], dependencies=[Depends(get_current_active_user)], responses={404: {"description": "User not found"}}, ) @router.get("/", response_model=List[UserSchema]) def read_users( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, current_user: User = Depends(get_current_active_user), ) -> Any: """ Retrieve users. """ users = get_users(db, skip=skip, limit=limit) return users @router.get("/{user_id}", response_model=UserSchema) def read_user( user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user), ) -> Any: """ Get a specific user by id. """ user = get_user(db, user_id=user_id) if user is None: raise HTTPException(status_code=404, detail="User not found") return user @router.patch("/{user_id}", response_model=UserSchema) def update_user_endpoint( user_id: int, user_in: UserUpdate, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user), ) -> Any: """ Update a user. """ # Only allow users to update their own information if current_user.id != user_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions to update this user" ) user = update_user(db, user_id=user_id, user=user_in) if user is None: raise HTTPException(status_code=404, detail="User not found") return user @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_user_endpoint( user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user), ) -> Any: """ Delete a user. """ # Only allow users to delete their own account if current_user.id != user_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions to delete this user" ) success = delete_user(db, user_id=user_id) if not success: raise HTTPException(status_code=404, detail="User not found") return None