Python: Backup MySQL Database

# ============================================================
# REAL-WORLD SCENARIOS
# ============================================================

print("""
1. Human Error
   - Developer accidentally deletes a table
   - User runs wrong UPDATE statement
   - Wrong data is inserted

2. System Failures
   - Server crashes
   - Hard disk fails
   - Power outages

3. Security Issues
   - Data breaches
   - Ransomware attacks
   - SQL injection

4. Application Bugs
   - Bug corrupts data
   - Migration goes wrong
   - Data inconsistency

5. Natural Disasters
   - Fire, flood, earthquake
   - Hardware destruction
   - Data center issues

💀 ANY of these can happen to you!
✅ A backup protects you from ALL of them!
""")

# ============================================================
# MYSQLDUMP BASIC COMMANDS
# ============================================================

# 1. Backup a single database
mysqldump -u root -p myapp_db > backup.sql

# 2. Backup with username and password
mysqldump -u root -psecret myapp_db > backup.sql

# 3. Backup multiple databases
mysqldump -u root -p --databases db1 db2 db3 > backup.sql

# 4. Backup all databases
mysqldump -u root -p --all-databases > backup.sql

# 5. Backup with compression
mysqldump -u root -p myapp_db | gzip > backup.sql.gz

# 6. Backup specific tables
mysqldump -u root -p myapp_db table1 table2 > backup.sql

# 7. Backup only structure (no data)
mysqldump -u root -p --no-data myapp_db > structure.sql

# 8. Backup only data (no structure)
mysqldump -u root -p --no-create-info myapp_db > data.sql

# 9. Backup with timestamp
mysqldump -u root -p myapp_db > backup_$(date +%Y%m%d).sql

# ============================================================
# RUNNING MYSQLDUMP FROM PYTHON
# ============================================================

import subprocess
import datetime

def backup_database(db_name, user, password, backup_dir):
    """Backup a MySQL database using mysqldump"""
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"{backup_dir}/{db_name}_{timestamp}.sql"
    
    command = f"mysqldump -u {user} -p{password} {db_name} > {filename}"
    
    try:
        subprocess.run(command, shell=True, check=True)
        print(f"✅ Backup created: {filename}")
        return filename
    except subprocess.CalledProcessError as e:
        print(f"❌ Backup failed: {e}")
        return None

# Usage
backup_database("myapp_db", "root", "secret", "/backups")

# ============================================================
# COMPLETE PYTHON BACKUP SCRIPT
# ============================================================

import subprocess
import os
import datetime
import gzip
import shutil
import logging

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('backup.log'),
        logging.StreamHandler()
    ]
)

class MySQLBackup:
    """MySQL database backup utility"""
    
    def __init__(self, host, user, password, backup_dir):
        self.host = host
        self.user = user
        self.password = password
        self.backup_dir = backup_dir
        self.logger = logging.getLogger(__name__)
        
        # Create backup directory if it doesn't exist
        os.makedirs(backup_dir, exist_ok=True)
    
    def backup_database(self, db_name, compress=True, keep_days=7):
        """Backup a single database"""
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{db_name}_{timestamp}.sql"
        filepath = os.path.join(self.backup_dir, filename)
        
        # Build mysqldump command
        cmd = [
            "mysqldump",
            f"--host={self.host}",
            f"--user={self.user}",
            f"--password={self.password}",
            "--single-transaction",  # For InnoDB consistency
            "--routines",  # Include stored procedures
            "--triggers",  # Include triggers
            db_name
        ]
        
        try:
            self.logger.info(f"Starting backup of {db_name}")
            
            # Run mysqldump
            with open(filepath, 'w') as f:
                subprocess.run(cmd, stdout=f, check=True)
            
            # Compress if requested
            if compress:
                compressed_path = filepath + '.gz'
                with open(filepath, 'rb') as f_in:
                    with gzip.open(compressed_path, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)
                os.remove(filepath)  # Remove uncompressed
                filepath = compressed_path
            
            self.logger.info(f"✅ Backup created: {filepath}")
            
            # Clean old backups
            self._clean_old_backups(db_name, keep_days)
            
            return filepath
            
        except subprocess.CalledProcessError as e:
            self.logger.error(f"❌ Backup failed: {e}")
            if os.path.exists(filepath):
                os.remove(filepath)
            return None
        except Exception as e:
            self.logger.error(f"❌ Error: {e}")
            return None
    
    def backup_multiple(self, db_names, compress=True, keep_days=7):
        """Backup multiple databases"""
        results = {}
        for db_name in db_names:
            results[db_name] = self.backup_database(db_name, compress, keep_days)
        return results
    
    def backup_all_databases(self, compress=True, keep_days=7):
        """Backup all databases"""
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"all_databases_{timestamp}.sql"
        filepath = os.path.join(self.backup_dir, filename)
        
        cmd = [
            "mysqldump",
            f"--host={self.host}",
            f"--user={self.user}",
            f"--password={self.password}",
            "--all-databases",
            "--single-transaction",
            "--routines",
            "--triggers"
        ]
        
        try:
            self.logger.info("Starting backup of all databases")
            
            with open(filepath, 'w') as f:
                subprocess.run(cmd, stdout=f, check=True)
            
            if compress:
                compressed_path = filepath + '.gz'
                with open(filepath, 'rb') as f_in:
                    with gzip.open(compressed_path, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)
                os.remove(filepath)
                filepath = compressed_path
            
            self.logger.info(f"✅ Full backup created: {filepath}")
            
            # Clean old backups
            self._clean_old_backups("all_databases", keep_days)
            
            return filepath
            
        except Exception as e:
            self.logger.error(f"❌ Backup failed: {e}")
            return None
    
    def _clean_old_backups(self, prefix, keep_days):
        """Delete old backup files"""
        try:
            now = datetime.datetime.now()
            cutoff = now - datetime.timedelta(days=keep_days)
            
            for filename in os.listdir(self.backup_dir):
                if filename.startswith(prefix):
                    filepath = os.path.join(self.backup_dir, filename)
                    mtime = datetime.datetime.fromtimestamp(os.path.getmtime(filepath))
                    
                    if mtime < cutoff:
                        os.remove(filepath)
                        self.logger.info(f"🗑️ Removed old backup: {filename}")
                        
        except Exception as e:
            self.logger.warning(f"Failed to clean old backups: {e}")
    
    def get_backup_size(self, filepath):
        """Get the size of a backup file"""
        if os.path.exists(filepath):
            size = os.path.getsize(filepath)
            if size < 1024:
                return f"{size} B"
            elif size < 1024 * 1024:
                return f"{size / 1024:.2f} KB"
            else:
                return f"{size / (1024 * 1024):.2f} MB"
        return "File not found"
    
    def list_backups(self):
        """List all backup files"""
        backups = []
        for filename in os.listdir(self.backup_dir):
            if filename.endswith('.sql') or filename.endswith('.sql.gz'):
                filepath = os.path.join(self.backup_dir, filename)
                mtime = datetime.datetime.fromtimestamp(os.path.getmtime(filepath))
                backups.append({
                    'name': filename,
                    'path': filepath,
                    'size': self.get_backup_size(filepath),
                    'modified': mtime
                })
        return sorted(backups, key=lambda x: x['modified'], reverse=True)

# ============================================================
# USAGE EXAMPLE
# ============================================================

if __name__ == "__main__":
    backup = MySQLBackup(
        host="localhost",
        user="root",
        password="secret",
        backup_dir="./backups"
    )
    
    # Backup a single database
    backup.backup_database("myapp_db")
    
    # Backup multiple databases
    backup.backup_multiple(["myapp_db", "test_db"])
    
    # Backup all databases
    backup.backup_all_databases()
    
    # List backups
    print("\n📋 Available backups:")
    for b in backup.list_backups()[:5]:
        print(f"  {b['name']} - {b['size']} - {b['modified']}")

# ============================================================
# AUTOMATED BACKUP SCHEDULER
# ============================================================

import schedule
import time
import smtplib
from email.mime.text import MIMEText

class BackupScheduler:
    """Schedule and automate database backups"""
    
    def __init__(self, backup_utility, email_config=None):
        self.backup = backup_utility
        self.email_config = email_config
    
    def daily_backup(self):
        """Run daily backup"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Starting daily backup...")
        result = self.backup.backup_all_databases()
        
        if result:
            print(f"✅ Daily backup completed: {result}")
            self._send_notification("Daily backup successful", result)
        else:
            print("❌ Daily backup failed")
            self._send_notification("Daily backup FAILED", "Check logs for details")
    
    def weekly_backup(self):
        """Run weekly full backup with verification"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Starting weekly backup...")
        
        # Backup all databases
        result = self.backup.backup_all_databases()
        
        if result:
            # Verify backup size
            size = self.backup.get_backup_size(result)
            print(f"✅ Weekly backup completed: {result} ({size})")
            self._send_notification("Weekly backup successful", f"{result} ({size})")
        else:
            print("❌ Weekly backup failed")
            self._send_notification("Weekly backup FAILED", "Check logs for details")
    
    def _send_notification(self, subject, message):
        """Send email notification"""
        if not self.email_config:
            return
        
        try:
            msg = MIMEText(message)
            msg['Subject'] = f"Backup Notification: {subject}"
            msg['From'] = self.email_config['from']
            msg['To'] = self.email_config['to']
            
            with smtplib.SMTP(self.email_config['smtp_server'], self.email_config['smtp_port']) as server:
                server.starttls()
                server.login(self.email_config['username'], self.email_config['password'])
                server.send_message(msg)
                
            print(f"📧 Notification sent: {subject}")
            
        except Exception as e:
            print(f"Failed to send notification: {e}")
    
    def run_scheduler(self):
        """Run the scheduler"""
        # Schedule backups
        schedule.every().day.at("02:00").do(self.daily_backup)
        schedule.every().sunday.at("03:00").do(self.weekly_backup)
        
        print("🔄 Backup scheduler started")
        print("📅 Daily backup at 2:00 AM")
        print("📅 Weekly backup on Sunday at 3:00 AM")
        print("Press Ctrl+C to stop")
        
        try:
            while True:
                schedule.run_pending()
                time.sleep(60)  # Check every minute
        except KeyboardInterrupt:
            print("\n⏹️ Scheduler stopped")

# ============================================================
# CONFIGURATION FILE
# ============================================================

# config.json
{
    "mysql": {
        "host": "localhost",
        "user": "root",
        "password": "secret"
    },
    "backup": {
        "directory": "/backups",
        "keep_days": 30
    },
    "email": {
        "smtp_server": "smtp.gmail.com",
        "smtp_port": 587,
        "username": "alerts@example.com",
        "password": "app_password",
        "from": "alerts@example.com",
        "to": "admin@example.com"
    }
}

# ============================================================
# RUNNING THE SCHEDULER
# ============================================================

if __name__ == "__main__":
    # Load config (simplified)
    backup = MySQLBackup("localhost", "root", "secret", "/backups")
    scheduler = BackupScheduler(backup)
    scheduler.run_scheduler()

# ============================================================
# RESTORING FROM BACKUP
# ============================================================

import subprocess
import gzip
import os

class MySQLRestore:
    """Restore MySQL databases from backups"""
    
    def __init__(self, host, user, password):
        self.host = host
        self.user = user
        self.password = password
    
    def restore_database(self, backup_file, db_name=None):
        """Restore a database from a backup file"""
        if not os.path.exists(backup_file):
            print(f"❌ Backup file not found: {backup_file}")
            return False
        
        try:
            # Determine if backup is compressed
            if backup_file.endswith('.gz'):
                # Decompress on the fly
                cmd = f"gunzip -c {backup_file} | mysql -u {self.user} -p{self.password}"
                if db_name:
                    cmd += f" {db_name}"
            else:
                # Direct restore
                cmd = f"mysql -u {self.user} -p{self.password}"
                if db_name:
                    cmd += f" {db_name}"
                cmd += f" < {backup_file}"
            
            print(f"🔄 Restoring from: {backup_file}")
            if db_name:
                print(f"📊 Database: {db_name}")
            
            subprocess.run(cmd, shell=True, check=True)
            print(f"✅ Restore successful!")
            return True
            
        except subprocess.CalledProcessError as e:
            print(f"❌ Restore failed: {e}")
            return False
        except Exception as e:
            print(f"❌ Error: {e}")
            return False
    
    def restore_with_confirmation(self, backup_file, db_name=None):
        """Restore with user confirmation"""
        print("\n⚠️ WARNING: This will overwrite the database!")
        print(f"📁 Backup: {backup_file}")
        if db_name:
            print(f"📊 Database: {db_name}")
        
        confirm = input("Are you sure you want to restore? (type 'YES' to proceed): ")
        if confirm == 'YES':
            return self.restore_database(backup_file, db_name)
        else:
            print("❌ Restore cancelled")
            return False

# ============================================================
# USAGE EXAMPLE
# ============================================================

restore = MySQLRestore("localhost", "root", "secret")

# Restore a database
restore.restore_database("/backups/myapp_db_20240115.sql", "myapp_db")

# Restore from compressed backup
restore.restore_database("/backups/all_databases_20240115.sql.gz")

# Restore with confirmation
restore.restore_with_confirmation("/backups/myapp_db_20240115.sql", "myapp_db")

# ============================================================
# PRODUCTION BACKUP SYSTEM
# ============================================================

import os
import datetime
import json
import logging
import subprocess
import shutil

class ProductionBackupSystem:
    """Complete backup system for production environments"""
    
    def __init__(self, config_file):
        self.config = self._load_config(config_file)
        self.setup_logging()
        self.logger = logging.getLogger(__name__)
    
    def _load_config(self, config_file):
        """Load configuration from JSON file"""
        with open(config_file, 'r') as f:
            return json.load(f)
    
    def setup_logging(self):
        """Set up logging configuration"""
        log_dir = self.config.get('logging', {}).get('directory', './logs')
        os.makedirs(log_dir, exist_ok=True)
        
        log_file = os.path.join(log_dir, f"backup_{datetime.datetime.now().strftime('%Y%m%d')}.log")
        
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()
            ]
        )
    
    def backup(self, backup_type='full'):
        """Run backup based on type"""
        self.logger.info(f"Starting {backup_type} backup")
        
        # Get backup directory
        backup_base = self.config['backup']['directory']
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        
        # Create backup directory
        if backup_type == 'full':
            backup_dir = os.path.join(backup_base, f"full_{timestamp}")
        else:
            backup_dir = os.path.join(backup_base, f"daily_{timestamp}")
        
        os.makedirs(backup_dir, exist_ok=True)
        
        try:
            # Backup each database
            for db_config in self.config['databases']:
                db_name = db_config['name']
                self._backup_database(db_name, backup_dir)
            
            # Create metadata
            self._create_metadata(backup_dir, backup_type)
            
            # Compress the entire backup
            self._compress_backup(backup_dir)
            
            # Clean old backups
            self._clean_old_backups(backup_base, backup_type)
            
            # Verify backup
            self._verify_backup(backup_dir + '.tar.gz')
            
            self.logger.info(f"✅ {backup_type} backup completed successfully")
            return True
            
        except Exception as e:
            self.logger.error(f"❌ {backup_type} backup failed: {e}")
            return False
    
    def _backup_database(self, db_name, backup_dir):
        """Backup a single database"""
        db_config = next(
            (db for db in self.config['databases'] if db['name'] == db_name),
            None
        )
        
        if not db_config:
            self.logger.warning(f"Database {db_name} not in config")
            return
        
        self.logger.info(f"📊 Backing up: {db_name}")
        
        # Build mysqldump command
        cmd = [
            "mysqldump",
            f"--host={self.config['mysql']['host']}",
            f"--user={self.config['mysql']['user']}",
            f"--password={self.config['mysql']['password']}",
            "--single-transaction",
            "--routines",
            "--triggers",
            db_name
        ]
        
        # Add options for specific databases
        if db_config.get('options'):
            cmd.extend(db_config['options'])
        
        # Output file
        backup_file = os.path.join(backup_dir, f"{db_name}.sql")
        
        with open(backup_file, 'w') as f:
            subprocess.run(cmd, stdout=f, check=True)
        
        self.logger.info(f"   ✅ {db_name} backed up to {backup_file}")
    
    def _create_metadata(self, backup_dir, backup_type):
        """Create metadata file for the backup"""
        metadata = {
            'backup_type': backup_type,
            'timestamp': datetime.datetime.now().isoformat(),
            'host': self.config['mysql']['host'],
            'databases': [db['name'] for db in self.config['databases']],
            'server_info': self._get_server_info()
        }
        
        metadata_file = os.path.join(backup_dir, 'metadata.json')
        with open(metadata_file, 'w') as f:
            json.dump(metadata, f, indent=2)
        
        self.logger.info(f"📋 Metadata created: {metadata_file}")
    
    def _get_server_info(self):
        """Get MySQL server information"""
        try:
            cmd = [
                "mysql",
                f"--host={self.config['mysql']['host']}",
                f"--user={self.config['mysql']['user']}",
                f"--password={self.config['mysql']['password']}",
                "-e", "SELECT VERSION()"
            ]
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            return {'version': result.stdout.strip().split('\n')[1]}
        except:
            return {'version': 'Unknown'}
    
    def _compress_backup(self, backup_dir):
        """Compress the backup directory"""
        import tarfile
        
        tar_file = backup_dir + '.tar.gz'
        self.logger.info(f"📦 Compressing backup: {tar_file}")
        
        with tarfile.open(tar_file, 'w:gz') as tar:
            tar.add(backup_dir, arcname=os.path.basename(backup_dir))
        
        # Remove uncompressed directory
        shutil.rmtree(backup_dir)
        
        self.logger.info(f"   ✅ Compressed to: {tar_file}")
        return tar_file
    
    def _clean_old_backups(self, backup_base, backup_type):
        """Clean old backups based on retention policy"""
        retention = self.config['backup'].get('retention', {})
        
        if backup_type == 'full':
            days = retention.get('full', 30)
            prefix = 'full_'
        else:
            days = retention.get('daily', 7)
            prefix = 'daily_'
        
        self.logger.info(f"🗑️ Cleaning backups older than {days} days")
        
        now = datetime.datetime.now()
        cutoff = now - datetime.timedelta(days=days)
        
        for item in os.listdir(backup_base):
            if item.startswith(prefix) and item.endswith('.tar.gz'):
                item_path = os.path.join(backup_base, item)
                mtime = datetime.datetime.fromtimestamp(os.path.getmtime(item_path))
                
                if mtime < cutoff:
                    os.remove(item_path)
                    self.logger.info(f"   🗑️ Removed old backup: {item}")
    
    def _verify_backup(self, backup_file):
        """Verify the backup file integrity"""
        self.logger.info(f"🔍 Verifying backup: {backup_file}")
        
        if not os.path.exists(backup_file):
            self.logger.error("❌ Backup file not found")
            return False
        
        # Check file size
        size = os.path.getsize(backup_file)
        if size < 1024:
            self.logger.error(f"❌ Backup file too small: {size} bytes")
            return False
        
        self.logger.info(f"   ✅ Backup verified: {size / (1024*1024):.2f} MB")
        return True
    
    def run(self):
        """Run the backup system"""
        print("\n" + "=" * 60)
        print("PRODUCTION BACKUP SYSTEM")
        print("=" * 60)
        
        # Run full backup weekly, daily backup otherwise
        if datetime.datetime.now().weekday() == 0:  # Monday
            self.backup('full')
        else:
            self.backup('daily')

# ============================================================
# CONFIGURATION FILE (config.json)
# ============================================================

# {
#     "mysql": {
#         "host": "localhost",
#         "user": "root",
#         "password": "secret"
#     },
#     "databases": [
#         {"name": "myapp_db", "options": ["--no-data"]},
#         {"name": "user_db"},
#         {"name": "order_db"}
#     ],
#     "backup": {
#         "directory": "/backups",
#         "retention": {
#             "daily": 7,
#             "full": 30
#         }
#     },
#     "logging": {
#         "directory": "/logs"
#     }
# }

# ============================================================
# RUNNING THE SYSTEM
# ============================================================

if __name__ == "__main__":
    backup_system = ProductionBackupSystem("config.json")
    backup_system.run()

# ============================================================
# BEST PRACTICES FOR DATABASE BACKUPS
# ============================================================

print("1. BACKUP REGULARLY")
print("   - Daily backups are a minimum")
print("   - Critical systems may need hourly backups")

print("\n2. STORE BACKUPS IN MULTIPLE LOCATIONS")
print("   - Keep backups on different drives")
print("   - Consider cloud storage as a backup location")
print("   - Follow the 3-2-1 rule:")
print("      - 3 copies of your data")
print("      - 2 different media types")
print("      - 1 copy offsite")

print("\n3. TEST YOUR BACKUPS")
print("   - Regularly test restoring from backups")
print("   - Verify that backups are not corrupted")
print("   - Document the restore process")

print("\n4. USE COMPRESSION FOR LARGE DATABASES")
print("   - Save storage space")
print("   - Faster transfer to offsite locations")

print("\n5. ENCRYPT SENSITIVE BACKUPS")
print("   - Protect customer data")
print("   - Use encryption for compliance")

print("\n6. MONITOR BACKUP SUCCESS")
print("   - Set up alerts for failed backups")
print("   - Check backup logs regularly")

print("\n7. ROTATE BACKUPS")
print("   - Keep daily backups for 7 days")
print("   - Keep weekly backups for 30 days")
print("   - Keep monthly backups for 1 year")

print("\n8. DOCUMENT YOUR BACKUP STRATEGY")
print("   - What gets backed up")
print("   - How often")
print("   - Where backups are stored")
print("   - How to restore")

print("\n9. BACKUP BEFORE MAJOR CHANGES")
print("   - Before running migrations")
print("   - Before software updates")
print("   - Before data modifications")

print("\n10. USE TRANSACTIONS FOR CONSISTENCY")
print("    - Use --single-transaction with mysqldump")
print("    - Ensures a consistent snapshot")