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 4.0 KiB

1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126
  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 core.config import Settings
  8. from db.models.user import User
  9. from schemas.user import UserCreate, ShowUser, ShowDriver, DriverCreate
  10. from db.session import get_db
  11. from db.repository.user import (
  12. create_new_user,
  13. list_users,
  14. get_user_by_id,
  15. replace_user_data,
  16. create_new_driver,
  17. delete_user_data,
  18. )
  19. router = APIRouter()
  20. @router.post("/", response_model=ShowUser, status_code=status.HTTP_201_CREATED)
  21. def create_user(
  22. user: UserCreate,
  23. db: Session = Depends(get_db),
  24. current_user: User = Depends(get_current_user),
  25. ):
  26. if user.Role not in Settings.ALLOWED_ROLES:
  27. raise HTTPException(
  28. status_code=400,
  29. detail=f"Status {status} is not allowed. Allowed status are {settings.ALLOWED_TASK_STATUS}",
  30. )
  31. if current_user.Role != "Admin":
  32. raise HTTPException(
  33. status_code=403, detail="You are not authorized to perform this action"
  34. )
  35. # if current_user.Role != "Admin":
  36. # raise HTTPException(status_code=403, detail="You are not authorized to perform this action")
  37. user = create_new_user(user=user, db=db)
  38. return user
  39. @router.post("/driver", response_model=ShowDriver, status_code=status.HTTP_201_CREATED)
  40. def create_driver(
  41. driver: DriverCreate,
  42. db: Session = Depends(get_db),
  43. current_user: User = Depends(get_current_user),
  44. ):
  45. if current_user.Role != "Admin":
  46. raise HTTPException(
  47. status_code=403, detail="You are not authorized to perform this action"
  48. )
  49. driver = create_new_driver(driver=driver, db=db)
  50. return driver
  51. @router.get("/", response_model=List[ShowUser], status_code=status.HTTP_200_OK)
  52. def get_all_users(db: Session = Depends(get_db), role: str = None):
  53. if role is None:
  54. users = list_users(db=db)
  55. return users
  56. users = list_users(db=db, role=role)
  57. return users
  58. @router.put("/{user_id}", response_model=ShowUser, status_code=status.HTTP_202_ACCEPTED)
  59. def update_user(
  60. user_id: int,
  61. user: UserCreate,
  62. db: Session = Depends(get_db),
  63. current_user: User = Depends(get_current_user),
  64. ):
  65. if current_user.Role != "Admin":
  66. raise HTTPException(
  67. status_code=403, detail="You are not authorized to perform this action"
  68. )
  69. user = replace_user_data(user_id=user_id, user=user, db=db)
  70. return user
  71. @router.get("/me", response_model=ShowUser, status_code=status.HTTP_200_OK)
  72. def get_user_me(
  73. current_user: Annotated[User, Depends(get_current_user)],
  74. db: Annotated[Session, Depends(get_db)],
  75. ):
  76. print("Getting current user...")
  77. return current_user
  78. @router.get("/{user_id}", response_model=ShowUser, status_code=status.HTTP_200_OK)
  79. def get_user(user_id: int, db: Session = Depends(get_db)):
  80. user = get_user_by_id(user_id=user_id, db=db)
  81. if not user:
  82. raise HTTPException(status_code=404, detail="User not found")
  83. return user
  84. @router.get(
  85. "/driver/{driver_id}", response_model=ShowDriver, status_code=status.HTTP_200_OK
  86. )
  87. def get_driver(driver_id: int, db: Session = Depends(get_db)):
  88. driver = get_user_by_id(user_id=driver_id, role="Driver", db=db)
  89. if not driver:
  90. raise HTTPException(status_code=404, detail="Driver not found")
  91. res = driver.__dict__
  92. res["AssignedVehicle"] = driver.vehicle
  93. return driver
  94. @router.delete("/{user_id}", status_code=status.HTTP_200_OK)
  95. def delete_user(
  96. user_id: int,
  97. db: Session = Depends(get_db),
  98. current_user: User = Depends(get_current_user),
  99. ):
  100. if current_user.Role != "Admin":
  101. raise HTTPException(
  102. status_code=403, detail="You are not authorized to perform this action"
  103. )
  104. result = delete_user_data(id=user_id, db=db)
  105. if result == "userNotFound":
  106. raise HTTPException(status_code=404, detail="User not found")
  107. return result