You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

route_user.py 2.3 KiB

1 vuosi sitten
1 vuosi sitten
1 vuosi sitten
1 vuosi sitten
1 vuosi sitten
1 vuosi sitten
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. # Routes for user. MAIN PART OF THE API
  2. from fastapi import APIRouter, HTTPException, status
  3. from sqlalchemy.orm import Session
  4. from fastapi import Depends
  5. from typing import List, Annotated
  6. from apis.v1.route_auth import get_current_user
  7. from db.models.user import User
  8. from schemas.user import UserCreate, ShowUser
  9. from db.session import get_db
  10. from db.repository.user import create_new_user, list_users, get_user_by_id, replace_user_data
  11. router = APIRouter()
  12. @router.post("/", response_model=ShowUser, status_code=status.HTTP_201_CREATED)
  13. def create_user(
  14. user: UserCreate,
  15. db: Session = Depends(get_db),
  16. current_user: User = Depends(get_current_user),
  17. ):
  18. if (current_user.Role != "Admin"):
  19. raise HTTPException(status_code=403, detail="You are not authorized to perform this action")
  20. # if current_user.Role != "Admin":
  21. # raise HTTPException(status_code=403, detail="You are not authorized to perform this action")
  22. user = create_new_user(user=user, db=db)
  23. return user
  24. @router.get("/", response_model=List[ShowUser], status_code=status.HTTP_200_OK)
  25. def get_all_users(db: Session = Depends(get_db), role: str = None):
  26. if role is None:
  27. users = list_users(db=db)
  28. return users
  29. users = list_users(db=db, role=role)
  30. return users
  31. @router.put("/{user_id}", response_model=ShowUser, status_code=status.HTTP_202_ACCEPTED)
  32. def update_user(
  33. user_id: int,
  34. user: UserCreate,
  35. db: Session = Depends(get_db),
  36. current_user: User = Depends(get_current_user),
  37. ):
  38. if current_user.Role != "Admin":
  39. raise HTTPException(status_code=403, detail="You are not authorized to perform this action")
  40. user = replace_user_data(user_id=user_id, user=user, db=db)
  41. return user
  42. @router.get("/me", response_model=ShowUser, status_code=status.HTTP_200_OK)
  43. def get_user_me(
  44. current_user: Annotated[User, Depends(get_current_user)],
  45. db: Annotated[Session, Depends(get_db)],
  46. ):
  47. print("Getting current user...")
  48. return current_user
  49. @router.get("/{user_id}", response_model=ShowUser, status_code=status.HTTP_200_OK)
  50. def get_user(user_id: int, db: Session = Depends(get_db)):
  51. user = get_user_by_id(user_id=user_id, db=db)
  52. if not user:
  53. raise HTTPException(status_code=404, detail="User not found")
  54. return user