Check the status of a backup/restore task. Requires either: - Valid admin authentication, OR - Valid task_token query parameter
(request, task_id)
| 72 | @api_view(["GET"]) |
| 73 | @permission_classes([AllowAny]) |
| 74 | def backup_status(request, task_id): |
| 75 | """Check the status of a backup/restore task. |
| 76 | |
| 77 | Requires either: |
| 78 | - Valid admin authentication, OR |
| 79 | - Valid task_token query parameter |
| 80 | """ |
| 81 | # Check for token-based auth (for restore when session is invalidated) |
| 82 | token = request.query_params.get("token") |
| 83 | if token: |
| 84 | if not _verify_task_token(task_id, token): |
| 85 | return Response( |
| 86 | {"detail": "Invalid task token"}, |
| 87 | status=status.HTTP_403_FORBIDDEN, |
| 88 | ) |
| 89 | else: |
| 90 | # Fall back to admin auth check |
| 91 | if not request.user.is_authenticated or getattr(request.user, 'user_level', 0) < 10: |
| 92 | return Response( |
| 93 | {"detail": "Authentication required"}, |
| 94 | status=status.HTTP_401_UNAUTHORIZED, |
| 95 | ) |
| 96 | |
| 97 | try: |
| 98 | result = AsyncResult(task_id) |
| 99 | |
| 100 | if result.ready(): |
| 101 | task_result = result.get() |
| 102 | if task_result.get("status") == "completed": |
| 103 | return Response({ |
| 104 | "state": "completed", |
| 105 | "result": task_result, |
| 106 | }) |
| 107 | else: |
| 108 | return Response({ |
| 109 | "state": "failed", |
| 110 | "error": task_result.get("error", "Unknown error"), |
| 111 | }) |
| 112 | elif result.failed(): |
| 113 | return Response({ |
| 114 | "state": "failed", |
| 115 | "error": str(result.result), |
| 116 | }) |
| 117 | else: |
| 118 | return Response({ |
| 119 | "state": result.state.lower(), |
| 120 | }) |
| 121 | except Exception as e: |
| 122 | return Response( |
| 123 | {"detail": f"Failed to get task status: {str(e)}"}, |
| 124 | status=status.HTTP_500_INTERNAL_SERVER_ERROR, |
| 125 | ) |
| 126 | |
| 127 | |
| 128 | @api_view(["GET"]) |
nothing calls this directly
no test coverage detected