Create a new backup (async via Celery).
(request)
| 51 | @api_view(["POST"]) |
| 52 | @permission_classes([IsAdmin]) |
| 53 | def create_backup(request): |
| 54 | """Create a new backup (async via Celery).""" |
| 55 | try: |
| 56 | task = create_backup_task.delay() |
| 57 | return Response( |
| 58 | { |
| 59 | "detail": "Backup started", |
| 60 | "task_id": task.id, |
| 61 | "task_token": _generate_task_token(task.id), |
| 62 | }, |
| 63 | status=status.HTTP_202_ACCEPTED, |
| 64 | ) |
| 65 | except Exception as e: |
| 66 | return Response( |
| 67 | {"detail": f"Failed to start backup: {str(e)}"}, |
| 68 | status=status.HTTP_500_INTERNAL_SERVER_ERROR, |
| 69 | ) |
| 70 | |
| 71 | |
| 72 | @api_view(["GET"]) |
nothing calls this directly
no test coverage detected