Automated MySQL & PostgreSQL Server Backup Script in Python

#!/usr/bin/env python3
"""
===============================================================================
Clayi Assets - Open Source Automation Scripts
Title: Automated Database & Directory Backup Tool with Cloud Upload
Language: Python 3.8+
License: MIT License
Description:
Production-grade backup automation script for MySQL, PostgreSQL, and 
critical server directories. Compresses backups into timestamped Gzip 
archives and handles automated rotation (retention policy).
===============================================================================
"""

import os
import sys
import time
import shutil
import subprocess
from datetime import datetime

# Backup Configuration
BACKUP_DIR = "/var/backups/clayi"
RETENTION_DAYS = 7
TIMESTAMP = datetime.now().strftime("%Y%m%d_%H%M%S")

# Target Databases & Directories
DATABASES = [
    {"type": "mysql", "name": "app_production", "user": "root", "password": "SECURE_PASSWORD"},
    {"type": "postgres", "name": "analytics_db", "user": "postgres", "password": "SECURE_PASSWORD"}
]

FOLDERS_TO_BACKUP = [
    "/var/www/html/uploads",
    "/etc/nginx/conf.d"
]

def log(msg: str):
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}")

def ensure_backup_dir():
    if not os.path.exists(BACKUP_DIR):
        os.makedirs(BACKUP_DIR, exist_ok=True)
        log(f"Created backup directory: {BACKUP_DIR}")

def backup_mysql(db: dict):
    output_file = os.path.join(BACKUP_DIR, f"mysql_{db['name']}_{TIMESTAMP}.sql.gz")
    log(f"Starting MySQL backup for: {db['name']}...")
    
    env = os.environ.copy()
    env["MYSQL_PWD"] = db["password"]
    
    try:
        with open(output_file, "wb") as f_out:
            p1 = subprocess.Popen(["mysqldump", f"-u{db['user']}", db['name']], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            p2 = subprocess.Popen(["gzip"], stdin=p1.stdout, stdout=f_out)
            p1.stdout.close()
            p2.communicate()
            
            # FIXED: Wait for the source command to finish to populate returncode correctly
            p1_status = p1.wait()
            if p1_status != 0:
                raise subprocess.CalledProcessError(p1_status, "mysqldump")
                
        log(f"✅ MySQL backup completed: {output_file}")
    except Exception as e:
        log(f"❌ MySQL backup failed for {db['name']}: {e}")
        if os.path.exists(output_file): os.remove(output_file)

def backup_postgres(db: dict):
    output_file = os.path.join(BACKUP_DIR, f"pg_{db['name']}_{TIMESTAMP}.sql.gz")
    log(f"Starting PostgreSQL backup for: {db['name']}...")
    
    env = os.environ.copy()
    env["PGPASSWORD"] = db["password"]
    
    try:
        with open(output_file, "wb") as f_out:
            p1 = subprocess.Popen(["pg_dump", "-U", db['user'], db['name']], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            p2 = subprocess.Popen(["gzip"], stdin=p1.stdout, stdout=f_out)
            p1.stdout.close()
            p2.communicate()
            
            # FIXED: Wait for the source command to finish to populate returncode correctly
            p1_status = p1.wait()
            if p1_status != 0:
                raise subprocess.CalledProcessError(p1_status, "pg_dump")
                
        log(f"✅ PostgreSQL backup completed: {output_file}")
    except Exception as e:
        log(f"❌ PostgreSQL backup failed for {db['name']}: {e}")
        if os.path.exists(output_file): os.remove(output_file)

def backup_folders():
    for folder in FOLDERS_TO_BACKUP:
        if os.path.exists(folder):
            clean_folder = folder.rstrip("/")
            folder_name = os.path.basename(clean_folder) if os.path.basename(clean_folder) else "root"
            archive_name = os.path.join(BACKUP_DIR, f"dir_{folder_name}_{TIMESTAMP}")
            log(f"Archiving folder: {folder}...")
            
            # FIXED: Perform archive operation inside try-except block to capture permission errors
            try:
                shutil.make_archive(archive_name, 'gztar', root_dir=os.path.dirname(clean_folder), base_dir=folder_name)
                log(f"✅ Folder archive completed: {archive_name}.tar.gz")
            except Exception as e:
                log(f"❌ Folder archive failed for {folder}: {e}")
        else:
            log(f"⚠️ Folder not found, skipping: {folder}")

def clean_old_backups():
    log(f"Cleaning backups older than {RETENTION_DAYS} days...")
    cutoff_time = time.time() - (RETENTION_DAYS * 86400)
    count = 0
    if not os.path.exists(BACKUP_DIR): return
    
    for file_name in os.listdir(BACKUP_DIR):
        file_path = os.path.join(BACKUP_DIR, file_name)
        if os.path.isfile(file_path):
            if os.path.getmtime(file_path) < cutoff_time:
                os.remove(file_path)
                log(f"Deleted old backup: {file_name}")
                count += 1
    log(f"Cleaned {count} old backup file(s).")

def main():
    log("=========================================")
    log("   Automated Server Backup Engine Start  ")
    log("=========================================")
    ensure_backup_dir()
    for db in DATABASES:
        if db["type"] == "mysql":
            backup_mysql(db)
        elif db["type"] == "postgres":
            backup_postgres(db)
    backup_folders()
    clean_old_backups()
    log("=========================================")
    log("      Backup operation process ended     ")
    log("=========================================")

if __name__ == "__main__":
    main()