Backup, Recovery & Disaster Planning

# PostgreSQL Full Backup with pg_dump
import subprocess
from datetime import datetime

def full_backup_postgres(db_name, backup_dir):
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_file = f"{backup_dir}/{db_name}_full_{timestamp}.sql"

    # Execute pg_dump
    cmd = [
        'pg_dump',
        '-h', 'localhost',
        '-U', 'postgres',
        '-d', db_name,
        '-F', 'c',  # Custom format (compressed)
        '-f', backup_file
    ]

    result = subprocess.run(cmd, capture_output=True, text=True)

    if result.returncode == 0:
        print(f"✅ Full backup completed: {backup_file}")
        # Get file size
        import os
        size_mb = os.path.getsize(backup_file) / (1024 * 1024)
        print(f"   Size: {size_mb:.2f} MB")
        return backup_file
    else:
        print(f"❌ Backup failed: {result.stderr}")
        return None

# Usage
backup_file = full_backup_postgres('mydb', '/backups')

# PostgreSQL WAL-based incremental backup
def incremental_backup_postgres(db_name, wal_archive_dir, backup_dir):
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')

    # Archive WAL files (Write-Ahead Logs)
    # These contain all changes since last backup
    cmd = [
        'pg_basebackup',
        '-h', 'localhost',
        '-U', 'postgres',
        '-D', f"{backup_dir}/incremental_{timestamp}",
        '-Ft',  # Tar format
        '-z',   # Compress
        '-P',   # Progress
        '--wal-method=fetch'
    ]

    result = subprocess.run(cmd, capture_output=True, text=True)

    if result.returncode == 0:
        print(f"✅ Incremental backup completed")
        print(f"   WAL files archived to: {wal_archive_dir}")
        return True
    else:
        print(f"❌ Backup failed: {result.stderr}")
        return False

# Usage
incremental_backup_postgres('mydb', '/wal_archive', '/backups')

# postgresql.conf settings for PITR

# Enable WAL archiving
wal_level = replica              # Generate enough WAL for PITR
archive_mode = on                # Enable archiving
archive_command = 'cp %p /wal_archive/%f'  # Copy WAL to archive dir

# WAL retention (keep enough for recovery)
wal_keep_size = 1GB              # Keep at least 1GB of WAL
max_wal_senders = 3              # Allow 3 streaming replicas

# Checkpoint settings (affects recovery time)
checkpoint_timeout = 5min        # Checkpoint every 5 minutes
max_wal_size = 1GB

# After changing config, reload PostgreSQL:
# sudo systemctl reload postgresql

# Python script to create PITR base backup
import subprocess
from datetime import datetime

def create_pitr_base_backup(backup_dir):
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_path = f"{backup_dir}/base_{timestamp}"

    # Create base backup with WAL files
    cmd = [
        'pg_basebackup',
        '-h', 'localhost',
        '-U', 'postgres',
        '-D', backup_path,
        '-Ft',              # Tar format
        '-z',               # Compress
        '-P',               # Show progress
        '--wal-method=stream',  # Include WAL files
        '--checkpoint=fast'     # Force immediate checkpoint
    ]

    print(f"Creating PITR base backup...")
    result = subprocess.run(cmd, capture_output=True, text=True)

    if result.returncode == 0:
        print(f"✅ Base backup created: {backup_path}")
        print(f"   Includes WAL files for PITR")
        return backup_path
    else:
        print(f"❌ Backup failed: {result.stderr}")
        return None

# Usage
backup_path = create_pitr_base_backup('/backups')

# Python script to perform PITR
import subprocess
import shutil
from datetime import datetime

def perform_pitr(base_backup_path, wal_archive_dir, target_time, restore_dir):
    """
    Restore database to specific point in time

    Args:
        base_backup_path: Path to base backup
        wal_archive_dir: Directory with archived WAL files
        target_time: Timestamp to restore to (e.g., '2025-02-05 14:30:00')
        restore_dir: Where to restore database
    """

    print(f"Starting PITR to {target_time}...")

    # 1. Extract base backup
    print("Step 1: Extracting base backup...")
    subprocess.run(['tar', '-xzf', f"{base_backup_path}/base.tar.gz", '-C', restore_dir])

    # 2. Create recovery configuration
    recovery_conf = f"""
restore_command = 'cp {wal_archive_dir}/%f %p'
recovery_target_time = '{target_time}'
recovery_target_action = 'promote'
    """

    with open(f"{restore_dir}/recovery.signal", 'w') as f:
        f.write("")  # Signal file for recovery mode

    with open(f"{restore_dir}/postgresql.auto.conf", 'a') as f:
        f.write(recovery_conf)

    print(f"✅ PITR configuration created")
    print(f"   Target time: {target_time}")
    print(f"   Restore directory: {restore_dir}")
    print(f"\n   Start PostgreSQL with this data directory to complete recovery")

    return restore_dir

# Usage: Restore to 1 minute before disaster
target_time = '2025-02-05 15:46:00'
restore_dir = perform_pitr(
    base_backup_path='/backups/base_20250205_140000',
    wal_archive_dir='/wal_archive',
    target_time=target_time,
    restore_dir='/var/lib/postgresql/14/main_restored'
)

# Automated backup verification script
import subprocess
import psycopg2
from datetime import datetime

def test_backup_restore(backup_file, test_db_name="test_restore_db"):
    """
    Test backup by actually restoring it and verifying data
    """
    print(f"Testing backup: {backup_file}")
    start_time = datetime.now()

    # Step 1: Drop test database if exists
    print("Step 1: Cleaning up old test database...")
    conn = psycopg2.connect("dbname=postgres user=postgres")
    conn.autocommit = True
    cur = conn.cursor()

    cur.execute(f"DROP DATABASE IF EXISTS {test_db_name}")
    cur.execute(f"CREATE DATABASE {test_db_name}")
    conn.close()

    # Step 2: Restore backup to test database
    print("Step 2: Restoring backup...")
    cmd = [
        'pg_restore',
        '-h', 'localhost',
        '-U', 'postgres',
        '-d', test_db_name,
        '-v',  # Verbose
        backup_file
    ]

    result = subprocess.run(cmd, capture_output=True, text=True)

    if result.returncode != 0:
        print(f"❌ Restore failed: {result.stderr}")
        return False

    # Step 3: Verify data integrity
    print("Step 3: Verifying data...")
    conn = psycopg2.connect(f"dbname={test_db_name} user=postgres")
    cur = conn.cursor()

    # Check table counts
    cur.execute("""
        SELECT schemaname, tablename, COUNT(*)
        FROM pg_tables
        WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
        GROUP BY schemaname, tablename
    """)

    tables = cur.fetchall()
    print(f"   Found {len(tables)} tables")

    # Verify critical tables have data
    cur.execute("SELECT COUNT(*) FROM users")
    user_count = cur.fetchone()[0]
    print(f"   Users table: {user_count} rows")

    cur.execute("SELECT COUNT(*) FROM transactions")
    tx_count = cur.fetchone()[0]
    print(f"   Transactions table: {tx_count} rows")

    conn.close()

    # Step 4: Cleanup
    print("Step 4: Cleaning up test database...")
    conn = psycopg2.connect("dbname=postgres user=postgres")
    conn.autocommit = True
    cur = conn.cursor()
    cur.execute(f"DROP DATABASE {test_db_name}")
    conn.close()

    elapsed = (datetime.now() - start_time).total_seconds()
    print(f"✅ Backup test PASSED in {elapsed:.2f}s")
    print(f"   Backup is valid and restorable!")

    return True

# Usage: Run weekly backup verification
test_backup_restore('/backups/mydb_full_20250205_143022.sql')

# Setup streaming replication to DR region

# PRIMARY (US-East): postgresql.conf
wal_level = replica
max_wal_senders = 10
wal_keep_size = 1GB
hot_standby = on

# Create replication user on PRIMARY
CREATE USER replication_user WITH REPLICATION PASSWORD 'secure_password';

# Allow replication connection in pg_hba.conf
# host replication replication_user DR_REPLICA_IP/32 md5

# REPLICA (US-West): Setup streaming from primary
pg_basebackup -h PRIMARY_IP -D /var/lib/postgresql/data -U replication_user -P -R

# REPLICA: postgresql.auto.conf (created by -R flag)
primary_conninfo = 'host=PRIMARY_IP port=5432 user=replication_user password=secure_password'