(self, input: dict, request: Request)
| 14 | return False |
| 15 | |
| 16 | async def process(self, input: dict, request: Request) -> dict | Response: |
| 17 | # Handle file upload |
| 18 | if 'backup_file' not in request.files: |
| 19 | return {"success": False, "error": "No backup file provided"} |
| 20 | |
| 21 | backup_file: FileStorage = request.files['backup_file'] |
| 22 | if backup_file.filename == '': |
| 23 | return {"success": False, "error": "No file selected"} |
| 24 | |
| 25 | # Get restore patterns and options from form data |
| 26 | metadata_json = request.form.get('metadata', '{}') |
| 27 | overwrite_policy = request.form.get('overwrite_policy', 'overwrite') |
| 28 | clean_before_restore = request.form.get('clean_before_restore', 'false').lower() == 'true' |
| 29 | |
| 30 | try: |
| 31 | metadata = json.loads(metadata_json) |
| 32 | restore_include_patterns = metadata.get("include_patterns", []) |
| 33 | restore_exclude_patterns = metadata.get("exclude_patterns", []) |
| 34 | except json.JSONDecodeError: |
| 35 | return {"success": False, "error": "Invalid metadata JSON"} |
| 36 | |
| 37 | try: |
| 38 | backup_service = BackupService() |
| 39 | result = await backup_service.preview_restore( |
| 40 | backup_file=backup_file, |
| 41 | restore_include_patterns=restore_include_patterns, |
| 42 | restore_exclude_patterns=restore_exclude_patterns, |
| 43 | overwrite_policy=overwrite_policy, |
| 44 | clean_before_restore=clean_before_restore, |
| 45 | user_edited_metadata=metadata |
| 46 | ) |
| 47 | |
| 48 | return { |
| 49 | "success": True, |
| 50 | "files": result["files"], |
| 51 | "files_to_delete": result.get("files_to_delete", []), |
| 52 | "files_to_restore": result.get("files_to_restore", []), |
| 53 | "skipped_files": result["skipped_files"], |
| 54 | "total_count": result["total_count"], |
| 55 | "delete_count": result.get("delete_count", 0), |
| 56 | "restore_count": result.get("restore_count", 0), |
| 57 | "skipped_count": result["skipped_count"], |
| 58 | "backup_metadata": result["backup_metadata"], |
| 59 | "overwrite_policy": result.get("overwrite_policy", "overwrite"), |
| 60 | "clean_before_restore": result.get("clean_before_restore", False) |
| 61 | } |
| 62 | |
| 63 | except Exception as e: |
| 64 | return { |
| 65 | "success": False, |
| 66 | "error": str(e) |
| 67 | } |
nothing calls this directly
no test coverage detected