Spaces:
Paused
Paused
File size: 2,274 Bytes
67e167f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
from typing import Annotated, List
from sqlalchemy.orm import Session
from fastapi import APIRouter, Depends, HTTPException
from app.db import get_db
from app.core import schemas, crud
from app.security import get_current_user
router = APIRouter()
@router.post("/create-user/", response_model=schemas.User)
def create_user(
user: schemas.UserCreate,
db: Annotated[Session, Depends(get_db)],
current_user: Annotated[schemas.User, Depends(get_current_user)],
):
if not current_user.is_superuser:
raise HTTPException(status_code=403, detail="Forbidden")
user_exists = crud.get_user_by_username(username=user.username, db=db)
if user_exists:
raise HTTPException(status_code=400, detail="Username already registered")
return crud.create_user(user=user, db=db)
@router.put("/update-user/", response_model=schemas.User)
def update_user(
user: schemas.UserUpdate,
db: Annotated[Session, Depends(get_db)],
current_user: Annotated[schemas.User, Depends(get_current_user)],
):
if not current_user.is_superuser:
raise HTTPException(status_code=403, detail="Forbidden")
user_exists = crud.get_user_by_user_id(user_id=user.user_id, db=db)
if not user_exists:
raise HTTPException(status_code=404, detail="User not found")
return crud.update_user(user=user, db=db)
@router.get("/get-all-users/", response_model=List[schemas.User])
def get_all_users(
db: Annotated[Session, Depends(get_db)],
current_user: Annotated[schemas.User, Depends(get_current_user)],
):
if not current_user.is_superuser:
raise HTTPException(status_code=403, detail="Forbidden")
return crud.get_all_users(db=db)
@router.get("/get-user_by_user_id/{user_id}/", response_model=schemas.User)
def get_user_by_user_id(
user_id: int,
db: Annotated[Session, Depends(get_db)],
current_user: Annotated[schemas.User, Depends(get_current_user)],
):
if not current_user.is_superuser:
raise HTTPException(status_code=403, detail="Forbidden")
user = crud.get_user_by_user_id(user_id=user_id, db=db)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
|