from typing import List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.db.session import get_db from app.crud import user from app.schemas.user import User, UserCreate, UserUpdate from app.auth.auth_handler import get_current_active_user from app.models.user import User as UserModel router = APIRouter() @router.post("/", response_model=User) def create_user( *, db: Session = Depends(get_db), user_in: UserCreate, current_user: UserModel = Depends(get_current_active_user) ): if not current_user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) db_user = user.get_by_email(db, email=user_in.email) if db_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="User with this email already exists" ) return user.create(db, obj_in=user_in) @router.get("/", response_model=List[User]) def read_users( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, current_user: UserModel = Depends(get_current_active_user) ): if not current_user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) return user.get_multi(db, skip=skip, limit=limit) @router.get("/me", response_model=User) def read_user_me( current_user: UserModel = Depends(get_current_active_user) ): return current_user @router.get("/{user_id}", response_model=User) def read_user( user_id: int, db: Session = Depends(get_db), current_user: UserModel = Depends(get_current_active_user) ): if not current_user.is_admin and current_user.id != user_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) db_user = user.get(db, id=user_id) if not db_user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return db_user @router.put("/{user_id}", response_model=User) def update_user( user_id: int, user_in: UserUpdate, db: Session = Depends(get_db), current_user: UserModel = Depends(get_current_active_user) ): if not current_user.is_admin and current_user.id != user_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) db_user = user.get(db, id=user_id) if not db_user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return user.update(db, db_obj=db_user, obj_in=user_in)