""" Backup Service This module provides backup functionality for MongoDB databases. Allows creating JSON backups of entire databases or specific collections. """ import os import json from datetime import datetime from typing import Dict, List, Any, Optional, Tuple from bson import ObjectId from services.mongo import MongoService class BackupService: """ Service for creating and managing MongoDB backups. Backups are stored as JSON files with metadata. """ def __init__(self, mongo_service: MongoService, backup_dir: str = "backups"): """ Initialize backup service. Args: mongo_service: MongoService instance backup_dir: Directory to store backups """ self.mongo = mongo_service self.backup_dir = backup_dir # Create backup directory if it doesn't exist if not os.path.exists(backup_dir): os.makedirs(backup_dir) def _convert_objectid_to_string(self, obj: Any) -> Any: """ Recursively convert ObjectId to string for JSON serialization. Args: obj: Object to convert Returns: Converted object """ if isinstance(obj, ObjectId): return str(obj) elif isinstance(obj, dict): return {key: self._convert_objectid_to_string(value) for key, value in obj.items()} elif isinstance(obj, list): return [self._convert_objectid_to_string(item) for item in obj] else: return obj def create_database_backup( self, database_name: str, include_collections: Optional[List[str]] = None, exclude_collections: Optional[List[str]] = None ) -> Tuple[bool, Optional[str], Optional[str]]: """ Create a backup of entire database or specific collections. Args: database_name: Name of the database to backup include_collections: Optional list of collections to include (None = all) exclude_collections: Optional list of collections to exclude Returns: Tuple of (success: bool, backup_path: Optional[str], error: Optional[str]) """ try: # Check connection if not self.mongo.is_connected(): return False, None, "MongoDB is not connected" # Get all collections in database all_collections = self.mongo.list_collections_in_database(database_name) if not all_collections: return False, None, f"No collections found in database '{database_name}'" # Determine which collections to backup if include_collections: collections_to_backup = [c for c in all_collections if c in include_collections] else: collections_to_backup = all_collections if exclude_collections: collections_to_backup = [c for c in collections_to_backup if c not in exclude_collections] if not collections_to_backup: return False, None, "No collections selected for backup" # Create backup data structure backup_data = { 'metadata': { 'database': database_name, 'timestamp': datetime.now().isoformat(), 'collections_count': len(collections_to_backup), 'collections': collections_to_backup }, 'data': {} } # Export each collection total_documents = 0 for collection_name in collections_to_backup: documents = self.mongo.export_collection_data(database_name, collection_name) # Convert ObjectId to string documents = [self._convert_objectid_to_string(doc) for doc in documents] backup_data['data'][collection_name] = { 'count': len(documents), 'documents': documents } total_documents += len(documents) # Update metadata with total count backup_data['metadata']['total_documents'] = total_documents # Generate backup filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"backup_{database_name}_{timestamp}.json" backup_path = os.path.join(self.backup_dir, backup_filename) # Save backup to file with open(backup_path, 'w', encoding='utf-8') as f: json.dump(backup_data, f, indent=2, ensure_ascii=False) return True, backup_path, None except Exception as e: return False, None, f"Backup failed: {str(e)}" def create_collection_backup( self, database_name: str, collection_name: str ) -> Tuple[bool, Optional[str], Optional[str]]: """ Create a backup of a single collection. Args: database_name: Name of the database collection_name: Name of the collection Returns: Tuple of (success: bool, backup_path: Optional[str], error: Optional[str]) """ return self.create_database_backup( database_name, include_collections=[collection_name] ) def list_backups(self) -> List[Dict[str, Any]]: """ List all available backups. Returns: List of backup metadata dictionaries """ backups = [] try: if not os.path.exists(self.backup_dir): return [] for filename in os.listdir(self.backup_dir): if filename.endswith('.json') and filename.startswith('backup_'): filepath = os.path.join(self.backup_dir, filename) # Get file stats stat = os.stat(filepath) # Try to read metadata try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) metadata = data.get('metadata', {}) except: metadata = {} backups.append({ 'filename': filename, 'filepath': filepath, 'size': stat.st_size, 'created': datetime.fromtimestamp(stat.st_mtime).isoformat(), 'database': metadata.get('database', 'Unknown'), 'collections_count': metadata.get('collections_count', 0), 'total_documents': metadata.get('total_documents', 0) }) # Sort by creation time (newest first) backups.sort(key=lambda x: x['created'], reverse=True) except Exception as e: print(f"Error listing backups: {e}") return backups def get_backup_info(self, backup_path: str) -> Optional[Dict[str, Any]]: """ Get detailed information about a backup file. Args: backup_path: Path to backup file Returns: Backup metadata or None if error """ try: with open(backup_path, 'r', encoding='utf-8') as f: data = json.load(f) return data.get('metadata') except Exception as e: print(f"Error reading backup info: {e}") return None def delete_backup(self, backup_path: str) -> Tuple[bool, Optional[str]]: """ Delete a backup file. Args: backup_path: Path to backup file Returns: Tuple of (success: bool, error: Optional[str]) """ try: if os.path.exists(backup_path): os.remove(backup_path) return True, None else: return False, "Backup file not found" except Exception as e: return False, f"Failed to delete backup: {str(e)}"