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, get_admin_user 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_admin_user)): return Users.get_users(skip, limit) ############################ # UpdateUserRole ############################ @router.post("/update/role", response_model=Optional[UserModel]) async def update_user_role(form_data: UserRoleUpdateForm, user=Depends(get_admin_user)): if user.id != form_data.id: return Users.update_user_role_by_id(form_data.id, form_data.role) raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACTION_PROHIBITED, ) ############################ # 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_admin_user) ): user = Users.get_user_by_id(user_id) if user: if form_data.email.lower() != user.email: email_user = Users.get_user_by_email(form_data.email.lower()) 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.lower()) updated_user = Users.update_user_by_id( user_id, { "name": form_data.name, "email": form_data.email.lower(), "profile_image_url": form_data.profile_image_url, }, ) if updated_user: return updated_user raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT(), ) 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_admin_user)): if user.id != user_id: result = Auths.delete_auth_by_id(user_id) if result: return True raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DELETE_USER_ERROR, ) raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACTION_PROHIBITED, )