RDS Snapshot Before Deployment & Point-in-Time Restore

# In your CI/CD pipeline (GitHub Actions, Jenkins, etc.)
python rds_snapshot.py pre-deploy --db prod-postgres --version v2.4.1

# Run your deployment...

# If deployment fails:
python rds_snapshot.py restore --snapshot pre-deploy-prod-postgres-v2.4.1-20250120-103045

import boto3
import time
import sys
from datetime import datetime, timezone


class RDSSnapshotManager:
    def __init__(self, region: str = "us-east-1"):
        """
        boto3.client("rds") is the RDS service client.
        RDS operations are long-running (snapshot: 5-30 min,
        restore: 10-45 min) — we use boto3 Waiters to poll for completion
        instead of writing manual sleep loops.
        """
        self.rds = boto3.client("rds", region_name=region)

    # ── Pre-deployment snapshot ───────────────────────────────────
    def create_pre_deploy_snapshot(
        self, db_identifier: str, deploy_version: str
    ) -> str:
        """
        create_db_snapshot() creates a manual snapshot of the RDS instance.
        Manual snapshots are retained until you explicitly delete them
        (unlike automated backups which expire based on the retention window).

        DBSnapshotIdentifier must be unique, 1-255 chars, alphanumeric + hyphens.
        We embed the instance name + deploy version + timestamp for traceability.

        TagSpecifications on RDS use a different pattern than EC2 — tags
        are passed directly in the API call.
        """
        timestamp   = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
        snapshot_id = f"pre-deploy-{db_identifier}-{deploy_version}-{timestamp}"

        print(f"Creating pre-deployment snapshot: {snapshot_id}")

        self.rds.create_db_snapshot(
            DBSnapshotIdentifier=snapshot_id,
            DBInstanceIdentifier=db_identifier,
            Tags=[
                {"Key": "Type",          "Value": "pre-deployment"},
                {"Key": "DeployVersion", "Value": deploy_version},
                {"Key": "CreatedAt",     "Value": timestamp},
                {"Key": "AutoDelete",    "Value": "false"},   # Keep until manually removed
            ],
        )

        # ── Wait for snapshot to be available ─────────────────────
        # get_waiter("db_snapshot_available") polls describe_db_snapshots()
        # every 30 seconds (default) until Status = "available".
        # Raises WaiterError if it doesn't complete within max_attempts.
        print("Waiting for snapshot to complete (5-20 min for large databases)...")
        waiter = self.rds.get_waiter("db_snapshot_available")
        waiter.wait(
            DBSnapshotIdentifier=snapshot_id,
            WaiterConfig={
                "Delay":       30,    # Poll every 30 seconds
                "MaxAttempts": 60,    # Give up after 30 min (60 × 30s)
            },
        )
        print(f"✅ Snapshot ready: {snapshot_id}")
        return snapshot_id

    # ── Restore from snapshot ─────────────────────────────────────
    def restore_from_snapshot(
        self,
        snapshot_id: str,
        new_db_identifier: str,
        db_subnet_group: str,
        vpc_sg_ids: list[str],
    ) -> str:
        """
        restore_db_instance_from_db_snapshot() creates a NEW RDS instance
        from the snapshot. It does NOT overwrite the existing database.

        This is intentional — you end up with TWO databases:
          prod-postgres         → current (potentially broken) DB
          prod-postgres-rollback → restored from pre-deploy snapshot

        After verifying the restore, you either:
          a) Swap the app config to point to the restored DB, OR
          b) Use pg_dump/restore to merge specific tables back

        We read DBInstanceClass from the snapshot so the restored instance
        matches the original instance type.
        """
        print(f"Restoring {new_db_identifier} from snapshot: {snapshot_id}")

        # Get the original instance class from the snapshot
        snapshots = self.rds.describe_db_snapshots(
            DBSnapshotIdentifier=snapshot_id
        )["DBSnapshots"]
        if not snapshots:
            raise ValueError(f"Snapshot not found: {snapshot_id}")
        snapshot = snapshots[0]

        self.rds.restore_db_instance_from_db_snapshot(
            DBInstanceIdentifier=new_db_identifier,
            DBSnapshotIdentifier=snapshot_id,
            # Use the same instance class as the original
            DBInstanceClass=snapshot["DBInstanceClass"],
            DBSubnetGroupName=db_subnet_group,
            VpcSecurityGroupIds=vpc_sg_ids,
            MultiAZ=True,
            AutoMinorVersionUpgrade=True,
            DeletionProtection=True,    # Prevent accidental deletion
            Tags=[
                {"Key": "RestoredFrom", "Value": snapshot_id},
                {"Key": "RestoredAt",   "Value": datetime.utcnow().isoformat() + "Z"},
                {"Key": "Temporary",    "Value": "true"},
            ],
        )

        print("Waiting for restored instance to be available (10-45 min)...")
        waiter = self.rds.get_waiter("db_instance_available")
        waiter.wait(
            DBInstanceIdentifier=new_db_identifier,
            WaiterConfig={"Delay": 30, "MaxAttempts": 120},   # Up to 60 min
        )

        endpoint = self.rds.describe_db_instances(
            DBInstanceIdentifier=new_db_identifier
        )["DBInstances"][0]["Endpoint"]["Address"]

        print(f"✅ Restored instance available at: {endpoint}")
        return endpoint

    # ── Point-in-time restore ─────────────────────────────────────
    def point_in_time_restore(
        self,
        source_db: str,
        target_db: str,
        restore_time: datetime,
        db_subnet_group: str,
        vpc_sg_ids: list[str],
    ) -> str:
        """
        restore_db_instance_to_point_in_time() restores to any second
        within the automated backup retention window (default 7 days).

        restore_time must be a timezone-aware datetime in UTC.
        Useful when you know exactly when the problem was introduced
        (e.g., "restore to 2025-01-20 14:29:59 UTC — 1 minute before deploy").

        Unlike snapshot restore, this uses the continuous transaction logs
        stored during automated backups to achieve precise RPO.
        """
        print(
            f"Restoring {source_db} to point-in-time: "
            f"{restore_time.strftime('%Y-%m-%d %H:%M:%S UTC')}"
        )

        self.rds.restore_db_instance_to_point_in_time(
            SourceDBInstanceIdentifier=source_db,
            TargetDBInstanceIdentifier=target_db,
            RestoreTime=restore_time,
            DBSubnetGroupName=db_subnet_group,
            VpcSecurityGroupIds=vpc_sg_ids,
            MultiAZ=True,
            DeletionProtection=True,
        )

        waiter = self.rds.get_waiter("db_instance_available")
        waiter.wait(
            DBInstanceIdentifier=target_db,
            WaiterConfig={"Delay": 30, "MaxAttempts": 120},
        )
        print(f"✅ Point-in-time restore complete: {target_db}")

        endpoint = self.rds.describe_db_instances(
            DBInstanceIdentifier=target_db
        )["DBInstances"][0]["Endpoint"]["Address"]
        return endpoint

    # ── List pre-deploy snapshots ─────────────────────────────────
    def list_pre_deploy_snapshots(self, db_identifier: str) -> list[dict]:
        """List all pre-deployment snapshots for a given DB, newest first."""
        paginator = self.rds.get_paginator("describe_db_snapshots")
        snapshots = []
        for page in paginator.paginate(
            DBInstanceIdentifier=db_identifier,
            SnapshotType="manual",
        ):
            for snap in page["DBSnapshots"]:
                if snap["DBSnapshotIdentifier"].startswith("pre-deploy-"):
                    snapshots.append({
                        "id":      snap["DBSnapshotIdentifier"],
                        "status":  snap["Status"],
                        "created": snap["SnapshotCreateTime"].strftime("%Y-%m-%d %H:%M UTC"),
                        "size_gb": snap.get("AllocatedStorage", 0),
                    })

        return sorted(snapshots, key=lambda x: x["created"], reverse=True)


# ── CLI entry point ───────────────────────────────────────────────
if __name__ == "__main__":
    manager = RDSSnapshotManager(region="ap-south-1")

    # Before deployment — called from CI/CD
    snapshot_id = manager.create_pre_deploy_snapshot(
        db_identifier="prod-postgres",
        deploy_version="v2.4.1",
    )
    print(f"\nSave this snapshot ID for rollback: {snapshot_id}")

    # List recent snapshots
    print("\nRecent pre-deploy snapshots:")
    for snap in manager.list_pre_deploy_snapshots("prod-postgres")[:5]:
        print(f"  {snap['id']}  [{snap['status']}]  {snap['created']}")

    # Rollback (uncomment if deployment fails):
    # manager.restore_from_snapshot(
    #     snapshot_id="pre-deploy-prod-postgres-v2.4.1-20250120-103045",
    #     new_db_identifier="prod-postgres-rollback",
    #     db_subnet_group="prod-db-subnet-group",
    #     vpc_sg_ids=["sg-0abc1234"],
    # )

    # Point-in-time restore (uncomment if needed):
    # from datetime import timezone
    # restore_to = datetime(2025, 1, 20, 14, 29, 59, tzinfo=timezone.utc)
    # manager.point_in_time_restore(
    #     source_db="prod-postgres",
    #     target_db="prod-postgres-pitr",
    #     restore_time=restore_to,
    #     db_subnet_group="prod-db-subnet-group",
    #     vpc_sg_ids=["sg-0abc1234"],
    # )

self.rds = boto3.client("rds", region_name=region)

timestamp   = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
snapshot_id = f"pre-deploy-{db_identifier}-{deploy_version}-{timestamp}"

self.rds.create_db_snapshot(
    DBSnapshotIdentifier=snapshot_id,
    DBInstanceIdentifier=db_identifier,
    Tags=[
        {"Key": "Type",          "Value": "pre-deployment"},
        {"Key": "DeployVersion", "Value": deploy_version},
    ],
)

waiter = self.rds.get_waiter("db_snapshot_available")
waiter.wait(
    DBSnapshotIdentifier=snapshot_id,
    WaiterConfig={"Delay": 30, "MaxAttempts": 60},
)

snapshots = self.rds.describe_db_snapshots(
    DBSnapshotIdentifier=snapshot_id
)["DBSnapshots"]
snapshot = snapshots[0]

self.rds.restore_db_instance_from_db_snapshot(
    DBInstanceIdentifier=new_db_identifier,
    DBSnapshotIdentifier=snapshot_id,
    DBInstanceClass=snapshot["DBInstanceClass"],
    DBSubnetGroupName=db_subnet_group,
    VpcSecurityGroupIds=vpc_sg_ids,
    MultiAZ=True,
    DeletionProtection=True,
)

endpoint = self.rds.describe_db_instances(
    DBInstanceIdentifier=new_db_identifier
)["DBInstances"][0]["Endpoint"]["Address"]

self.rds.restore_db_instance_to_point_in_time(
    SourceDBInstanceIdentifier=source_db,
    TargetDBInstanceIdentifier=target_db,
    RestoreTime=restore_time,
    ...
)

for page in paginator.paginate(
    DBInstanceIdentifier=db_identifier,
    SnapshotType="manual",
):
    for snap in page["DBSnapshots"]:
        if snap["DBSnapshotIdentifier"].startswith("pre-deploy-"):