77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
from fastapi.routing import APIRouter
|
|
from data.schemas import UserOut, UserInDBBase, UserCreate
|
|
from data.models import UserRole
|
|
from fastapi import FastAPI, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from data.crud import read_all_users, read_user, create_user, delete_user
|
|
from core.dependencies import get_db, get_current_user, get_admin_user
|
|
from sqlalchemy.exc import IntegrityError
|
|
from core.exceptions import ObjectNotFoundInDB, UserNotFound
|
|
from core.scripts import create_secret
|
|
|
|
users_router = APIRouter()
|
|
|
|
|
|
@users_router.get("/me")
|
|
async def get_me(user=Depends(get_current_user)) -> UserOut:
|
|
return user
|
|
|
|
|
|
@users_router.get(
|
|
"/",
|
|
dependencies=[Depends(get_current_user)],
|
|
)
|
|
async def get_all_users_endpoint(db=Depends(get_db)) -> list[UserOut]:
|
|
return await read_all_users(db=db)
|
|
|
|
|
|
@users_router.post(
|
|
"/", dependencies=[Depends(get_current_user)], status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def create_user_endpoint(
|
|
user_create: UserCreate, db=Depends(get_db)
|
|
) -> UserInDBBase:
|
|
try:
|
|
return await create_user(user=user_create, db=db)
|
|
except IntegrityError:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail={
|
|
"message": "This username is already taken.",
|
|
"code": "duplicated-username",
|
|
},
|
|
)
|
|
|
|
@users_router.post('/update-my-api_key/', status_code=status.HTTP_204_NO_CONTENT)
|
|
async def update_user_own_api_key(user=Depends(get_current_user), db=Depends(get_db)):
|
|
if user.role == UserRole.admin:
|
|
raise HTTPException(status_code=400, detail={
|
|
'message': 'Admins can\'t use this endpoint to update their API key.',
|
|
'code': 'admin-not-allowed'
|
|
})
|
|
user.api_key = create_secret()
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
|
|
@users_router.post('/update-user-api_key/', status_code=status.HTTP_202_ACCEPTED, dependencies=[Depends(get_admin_user)])
|
|
async def update_user_own_api_key(user_id:int, db=Depends(get_db)) -> UserInDBBase:
|
|
user = await read_user(db=db, user_id=user_id)
|
|
if user is None:
|
|
raise UserNotFound()
|
|
user.api_key = create_secret()
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
@users_router.delete(
|
|
"/", dependencies=[Depends(get_admin_user)], status_code=status.HTTP_204_NO_CONTENT
|
|
)
|
|
async def delete_user_endpoint(user_id: int, db=Depends(get_db)):
|
|
try:
|
|
await delete_user(db=db, user_id=user_id)
|
|
return None
|
|
except ObjectNotFoundInDB:
|
|
raise UserNotFound()
|