open-webui/backend/apps/web/routers/users.py

88 lines
2.4 KiB
Python
Raw Normal View History

2023-11-19 09:13:59 +01:00
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, UserRoleUpdateForm, Users
2023-12-29 08:24:51 +01:00
from apps.web.models.auths import Auths
2023-11-19 09:13:59 +01:00
from utils.utils import get_current_user
2023-11-19 09:13:59 +01:00
from constants import ERROR_MESSAGES
2023-11-19 09:13:59 +01:00
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":
2023-11-19 09:13:59 +01:00
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
2023-11-19 09:13:59 +01:00
)
return Users.get_users(skip, limit)
2023-11-19 09:13:59 +01:00
############################
# UpdateUserRole
############################
@router.post("/update/role", response_model=Optional[UserModel])
async def update_user_role(
form_data: UserRoleUpdateForm, user=Depends(get_current_user)
):
if user.role != "admin":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
)
2023-11-19 09:13:59 +01:00
if user.id != form_data.id:
return Users.update_user_role_by_id(form_data.id, form_data.role)
2023-11-19 09:13:59 +01:00
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=ERROR_MESSAGES.ACTION_PROHIBITED,
2023-11-19 09:13:59 +01:00
)
2023-12-29 08:02:49 +01:00
############################
2023-12-29 08:24:51 +01:00
# DeleteUserById
2023-12-29 08:02:49 +01:00
############################
@router.delete("/{user_id}", response_model=bool)
2023-12-30 12:00:21 +01:00
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
2023-12-29 08:02:49 +01:00
else:
raise HTTPException(
2023-12-30 12:00:21 +01:00
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=ERROR_MESSAGES.DELETE_USER_ERROR,
2023-12-29 08:02:49 +01:00
)
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
2023-12-30 12:00:21 +01:00
detail=ERROR_MESSAGES.ACTION_PROHIBITED,
2023-12-29 08:02:49 +01:00
)
else:
raise HTTPException(
2023-12-30 12:00:21 +01:00
status_code=status.HTTP_403_FORBIDDEN,
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
2023-12-29 08:02:49 +01:00
)