Restore from a backup file (async via Celery). WARNING: This will flush the database!
(request, filename)
| 311 | @api_view(["POST"]) |
| 312 | @permission_classes([IsAdmin]) |
| 313 | def restore_backup(request, filename): |
| 314 | """Restore from a backup file (async via Celery). WARNING: This will flush the database!""" |
| 315 | try: |
| 316 | # Security: prevent path traversal |
| 317 | if ".." in filename or "/" in filename or "\\" in filename: |
| 318 | raise Http404("Invalid filename") |
| 319 | |
| 320 | backup_dir = services.get_backup_dir() |
| 321 | backup_file = backup_dir / filename |
| 322 | |
| 323 | if not backup_file.exists(): |
| 324 | raise Http404("Backup file not found") |
| 325 | |
| 326 | task = restore_backup_task.delay(filename) |
| 327 | return Response( |
| 328 | { |
| 329 | "detail": "Restore started", |
| 330 | "task_id": task.id, |
| 331 | "task_token": _generate_task_token(task.id), |
| 332 | }, |
| 333 | status=status.HTTP_202_ACCEPTED, |
| 334 | ) |
| 335 | except Http404: |
| 336 | raise |
| 337 | except Exception as e: |
| 338 | return Response( |
| 339 | {"detail": f"Failed to start restore: {str(e)}"}, |
| 340 | status=status.HTTP_500_INTERNAL_SERVER_ERROR, |
| 341 | ) |
| 342 | |
| 343 | |
| 344 | @api_view(["GET"]) |
nothing calls this directly
no test coverage detected