from fastapi import Response from fastapi import Depends, FastAPI, HTTPException, status from datetime import datetime, timedelta from typing import List, Union, Optional from fastapi import APIRouter from pydantic import BaseModel import time import uuid from apps.web.models.users import UserModel, UserUpdateForm, UserRoleUpdateForm, Users from apps.web.models.auths import Auths from utils.utils import get_current_user, get_password_hash from constants import ERROR_MESSAGES router = APIRouter() ############################ # GetUsers ############################ @router.get("/", response_model=List[UserModel]) async def get_users(skip: int = 0, limit: int = 50, user=Depends(get_current_user)): if user.role != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) return Users.get_users(skip, limit) ############################ # UpdateUserById ############################ @router.post("/{user_id}/update", response_model=Optional[UserModel]) async def update_user_by_id( user_id: str, form_data: UserUpdateForm, session_user=Depends(get_current_user) ): if session_user.role != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) user = Users.get_user_by_id(user_id) if user: if form_data.email != user.email: email_user = Users.get_user_by_email(form_data.email) if email_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.EMAIL_TAKEN, ) if form_data.password: hashed = get_password_hash(form_data.password) print(hashed) Auths.update_user_password_by_id(user_id, hashed) Auths.update_email_by_id(user_id, form_data.email) updated_user = Users.update_user_by_id( user_id, { "name": form_data.name, "email": form_data.email, "profile_image_url": form_data.profile_image_url, }, ) if updated_user: return updated_user else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT(), ) else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.USER_NOT_FOUND, ) ############################ # DeleteUserById ############################ @router.delete("/{user_id}", response_model=bool) async def delete_user_by_id(user_id: str, user=Depends(get_current_user)): if user.role == "admin": if user.id != user_id: result = Auths.delete_auth_by_id(user_id) if result: return True else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DELETE_USER_ERROR, ) else: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACTION_PROHIBITED, ) else: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED, )