✨ Added admin-option to delete an account

classquiz/__init__.py

@@ -31,6 +31,7 @@ from classquiz.routers import (
     community,
     avatar,
     results,
+    admin,
 )
 from classquiz.socket_server import sio
 from classquiz.helpers import meilisearch_init, telemetry_ping, bg_tasks
@@ -104,5 +105,5 @@ app.include_router(eximport.router, tags=["export", "import"], prefix="/api/v1/e
 app.include_router(sitemap.router, tags=["sitemap"], prefix="/api/v1/sitemap", include_in_schema=True)
 app.include_router(community.router, tags=["community"], prefix="/api/v1/community", include_in_schema=True)
 app.include_router(avatar.router, tags=["avatar"], prefix="/api/v1/avatar", include_in_schema=True)
-
+app.include_router(admin.router, tags=["admin"], prefix="/api/v1/admin", include_in_schema=True)
 app.mount("/", ASGIApp(sio))

classquiz/auth.py

@@ -127,6 +127,15 @@ async def get_current_user(token: str = Depends(oauth2_scheme)):
     return user
 
 
+async def get_admin_user(token: str = Depends(oauth2_scheme)) -> User:
+    user = await get_current_user(token)
+    admin_user = await User.objects.order_by(User.created_at.asc()).get()
+    if admin_user.id == user.id:
+        return user
+    else:
+        raise credentials_exception
+
+
 async def get_current_user_optional(token: str = Depends(oauth2_scheme)) -> User | None:
     try:
         payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])

classquiz/routers/admin.py

@@ -0,0 +1,26 @@
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+
+from classquiz.auth import get_admin_user
+from classquiz.db.models import User
+
+router = APIRouter()
+
+
+@router.delete("/user/id")
+async def delete_user_by_id(user_id: UUID, user: User = Depends(get_admin_user)):
+    return {"deleted": await User.objects.delete(id=user_id)}
+
+
+@router.delete("/user/username")
+async def delete_user_by_username(username: str, user: User = Depends(get_admin_user)):
+    return {"deleted": await User.objects.delete(username=username)}
+
+
+@router.delete("/user/email")
+async def delete_user_by_email(email: str, user: User = Depends(get_admin_user)):
+    return {"deleted": await User.objects.delete(email=email)}