admin.py 1.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. from fastapi import APIRouter, Depends, HTTPException, Query
  2. from typing import List, Optional
  3. import db
  4. import auth_utils
  5. from dependencies import require_admin
  6. router = APIRouter(prefix="/admin", tags=["admin"])
  7. @router.get("/audit-logs")
  8. async def get_audit_logs(
  9. page: int = Query(1, ge=1),
  10. size: int = Query(50, ge=1, le=100),
  11. action: Optional[str] = None,
  12. admin: dict = Depends(require_admin)
  13. ):
  14. offset = (page - 1) * size
  15. query = """
  16. SELECT a.*, u.email as user_email
  17. FROM audit_logs a
  18. LEFT JOIN users u ON a.user_id = u.id
  19. """
  20. params = []
  21. if action:
  22. query += " WHERE a.action = %s"
  23. params.append(action)
  24. query += " ORDER BY a.created_at DESC LIMIT %s OFFSET %s"
  25. params.extend([size, offset])
  26. logs = db.execute_query(query, tuple(params))
  27. # Total count for pagination
  28. count_query = "SELECT COUNT(*) as total FROM audit_logs"
  29. if action:
  30. count_query += " WHERE action = %s"
  31. total = db.execute_query(count_query, (action,))
  32. else:
  33. total = db.execute_query(count_query)
  34. return {
  35. "logs": logs,
  36. "total": total[0]['total'],
  37. "page": page,
  38. "size": size
  39. }