2024-08-09 22:34:54 +03:00
|
|
|
import secrets
|
|
|
|
|
2024-08-10 17:01:34 +03:00
|
|
|
from fastapi import APIRouter, HTTPException
|
|
|
|
from sqlalchemy import select, delete
|
2024-08-09 22:34:54 +03:00
|
|
|
|
|
|
|
import database
|
2024-08-10 17:01:34 +03:00
|
|
|
from models import settings, user, DeleteUser
|
|
|
|
from .utils import Admin, hash_password
|
2024-08-09 22:34:54 +03:00
|
|
|
|
|
|
|
router = APIRouter(prefix="/admin")
|
|
|
|
|
|
|
|
|
|
|
|
@router.post("/user")
|
|
|
|
async def create_user(auth: user.Auth, admin_token: Admin):
|
|
|
|
if len(auth.username.strip()) == 0:
|
|
|
|
raise HTTPException(400, "Username must not be empty")
|
|
|
|
if len(auth.password.strip()) == 0:
|
|
|
|
raise HTTPException(400, "Password must not be empty")
|
|
|
|
if settings.disable_admin:
|
|
|
|
raise HTTPException(403, "You are not admin")
|
|
|
|
|
|
|
|
salt = secrets.token_hex(8)
|
|
|
|
|
|
|
|
async with database.sessions.begin() as session:
|
2024-08-10 17:01:34 +03:00
|
|
|
stmt = select(database.User).where(
|
|
|
|
database.User.username == auth.username.strip()
|
|
|
|
)
|
|
|
|
db_request = await session.execute(stmt)
|
|
|
|
user = db_request.scalar_one_or_none()
|
|
|
|
if user is not None:
|
2024-08-09 22:34:54 +03:00
|
|
|
raise HTTPException(400, "User with this username already exists")
|
|
|
|
|
|
|
|
new_user = database.User(
|
2024-08-10 17:01:34 +03:00
|
|
|
username=auth.username.strip(),
|
|
|
|
password=hash_password(auth.password.strip(), salt),
|
2024-08-09 22:34:54 +03:00
|
|
|
salt=salt,
|
|
|
|
)
|
|
|
|
session.add(new_user)
|
|
|
|
|
2024-08-10 17:01:34 +03:00
|
|
|
|
|
|
|
@router.delete("/user")
|
|
|
|
async def delete_user(user: DeleteUser, admin_token: Admin):
|
|
|
|
async with database.sessions.begin() as session:
|
|
|
|
stmt = delete(database.User).where(
|
|
|
|
database.User.username == user.username.strip()
|
|
|
|
)
|
|
|
|
await session.execute(stmt)
|