|
from typing import List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
from ..auth import get_current_admin_user, get_password_hash
|
|
from ..models import User, User_Pydantic
|
|
from ..schemas import UserCreate, UserAdminUpdate
|
|
|
|
router = APIRouter(
|
|
prefix="/admin",
|
|
tags=["admin"],
|
|
dependencies=[Depends(get_current_admin_user)],
|
|
responses={403: {"description": "Not enough permissions"}},
|
|
)
|
|
|
|
|
|
@router.get("/users", response_model=List[User_Pydantic])
|
|
async def get_all_users():
|
|
return await User.all()
|
|
|
|
|
|
@router.get("/users/{user_id}", response_model=User_Pydantic)
|
|
async def get_user(user_id: int):
|
|
user = await User.get_or_none(id=user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
|
|
)
|
|
return user
|
|
|
|
|
|
@router.post(
|
|
"/users", response_model=User_Pydantic, status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def create_user_by_admin(user_in: UserCreate):
|
|
user = await User.get_or_none(username=user_in.username)
|
|
if user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Username already registered",
|
|
)
|
|
user = await User.get_or_none(email=user_in.email)
|
|
if user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Email already registered",
|
|
)
|
|
|
|
hashed_password = get_password_hash(user_in.password)
|
|
user = await User.create(
|
|
username=user_in.username,
|
|
email=user_in.email,
|
|
hashed_password=hashed_password,
|
|
is_superuser=False, # Admin creates regular users by default
|
|
is_active=True,
|
|
)
|
|
return await User_Pydantic.from_tortoise_orm(user)
|
|
|
|
|
|
@router.put("/users/{user_id}", response_model=User_Pydantic)
|
|
async def update_user_by_admin(user_id: int, user_update: UserAdminUpdate):
|
|
user = await User.get_or_none(id=user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
|
|
)
|
|
|
|
if user_update.username is not None and user_update.username != user.username:
|
|
if await User.get_or_none(username=user_update.username):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="Username already taken"
|
|
)
|
|
user.username = user_update.username
|
|
|
|
if user_update.email is not None and user_update.email != user.email:
|
|
if await User.get_or_none(email=user_update.email):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Email already registered",
|
|
)
|
|
user.email = user_update.email
|
|
|
|
if user_update.password is not None:
|
|
user.hashed_password = get_password_hash(user_update.password)
|
|
|
|
if user_update.is_active is not None:
|
|
user.is_active = user_update.is_active
|
|
|
|
if user_update.is_superuser is not None:
|
|
user.is_superuser = user_update.is_superuser
|
|
|
|
if user_update.storage_quota_bytes is not None:
|
|
user.storage_quota_bytes = user_update.storage_quota_bytes
|
|
|
|
if user_update.plan_type is not None:
|
|
user.plan_type = user_update.plan_type
|
|
|
|
if user_update.is_2fa_enabled is not None:
|
|
user.is_2fa_enabled = user_update.is_2fa_enabled
|
|
if not user_update.is_2fa_enabled:
|
|
user.two_factor_secret = None
|
|
user.recovery_codes = None
|
|
|
|
await user.save()
|
|
return await User_Pydantic.from_tortoise_orm(user)
|
|
|
|
|
|
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_user_by_admin(user_id: int):
|
|
user = await User.get_or_none(id=user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
|
|
)
|
|
await user.delete()
|
|
return {"message": "User deleted successfully"}
|
|
|
|
|
|
@router.post("/test-email")
|
|
async def send_test_email(
|
|
to_email: str, subject: str = "Test Email", body: str = "This is a test email"
|
|
):
|
|
from ..mail import queue_email
|
|
|
|
queue_email(
|
|
to_email=to_email,
|
|
subject=subject,
|
|
body=body,
|
|
html=f"<h1>{subject}</h1><p>{body}</p>",
|
|
)
|
|
return {"message": "Test email queued"}
|