81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
# modules/admin/api.py
|
|
from typing import Annotated, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
from core.database import get_db
|
|
from modules.auth.dependencies import admin_only
|
|
from modules.auth.models import User
|
|
from modules.notifications.service import send_push_notification
|
|
from .tasks import cleardb
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"], dependencies=[Depends(admin_only)])
|
|
|
|
|
|
class ClearDbRequest(BaseModel):
|
|
hard: bool
|
|
|
|
|
|
class SendNotificationRequest(BaseModel):
|
|
username: str
|
|
title: str
|
|
body: str
|
|
data: Optional[dict] = None
|
|
|
|
|
|
@router.get("/")
|
|
def read_admin():
|
|
return {"message": "Admin route"}
|
|
|
|
|
|
@router.post("/cleardb")
|
|
def clear_db(payload: ClearDbRequest, db: Annotated[Session, Depends(get_db)]):
|
|
"""
|
|
Clear the database based on the 'hard' flag in the request body.
|
|
'hard'=True: Drop and recreate all tables.
|
|
'hard'=False: Delete data from tables except users.
|
|
"""
|
|
hard = payload.hard
|
|
cleardb.delay(hard)
|
|
return {"message": "Clearing database in the background", "hard": hard}
|
|
|
|
|
|
@router.post("/send-notification", status_code=status.HTTP_200_OK)
|
|
async def send_user_notification(
|
|
payload: SendNotificationRequest,
|
|
db: Annotated[Session, Depends(get_db)],
|
|
):
|
|
"""
|
|
Admin endpoint to send a push notification to a specific user by username.
|
|
"""
|
|
target_user = db.query(User).filter(User.username == payload.username).first()
|
|
|
|
if not target_user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"User with username '{payload.username}' not found.",
|
|
)
|
|
|
|
if not target_user.expo_push_token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"User '{payload.username}' does not have a registered push token.",
|
|
)
|
|
|
|
success = await send_push_notification(
|
|
push_token=target_user.expo_push_token,
|
|
title=payload.title,
|
|
body=payload.body,
|
|
data=payload.data,
|
|
)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to send push notification via Expo service.",
|
|
)
|
|
|
|
return {
|
|
"message": f"Push notification sent successfully to user '{payload.username}'"
|
|
}
|