Python code deleted wrong snapshots and tried to copy wrong snapshots

Hello.

The "describe_db_snapshots" API allows you to specify a DBInstanceIdentifier as shown below.
So, why not try filtering the snapshots to delete by specifying a DBInstanceIdentifier?
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rds/client/describe_db_snapshots.html

def delete_oldest_snapshots(dest_rds, count):

    db_names = [
        "demo", "drtest20240314"
    ]

# Get all manual snapshots in destination region
    for db_name in db_names:
        response = dest_rds.describe_db_snapshots(
            DBInstanceIdentifier=db_name,
            SnapshotType='manual')

    # Sort snapshots by creation time (oldest first)
        snapshots = sorted(response['DBSnapshots'], key=lambda s: s['SnapshotCreateTime'])

    # Delete the oldest 'count' snapshots
        for i, snapshot in enumerate(snapshots):
            if i >= count:
                break

            snapshot_id = snapshot['DBSnapshotIdentifier']
            print(f"Deleting snapshot: {snapshot_id}")

            try:
                dest_rds.delete_db_snapshot(DBSnapshotIdentifier=snapshot_id)
                print(f"Successfully deleted snapshot: {snapshot_id}")
            except Exception as e:
                print(f"Error deleting snapshot {snapshot_id}: {str(e)}")

Also, when copying a snapshot, please specify the KMS key for the destination region.
This can be specified using the KmsKeyId parameter of the "copy_db_snapshot" API.
The code has been modified so that when copying snapshots, only the DBs listed in the "db_names" list are included.
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rds/client/copy_db_cluster_snapshot.html

def copy_latest_snapshots(source_rds, dest_rds, count):
    # Define the database names based on your example
    #  "amgen", "alpine"
    db_names = [
        "demo", "drtest20240314"
    ]

    for db_name in db_names:
    # Get all automated snapshots in source region
        response = source_rds.describe_db_snapshots(
            DBInstanceIdentifier=db_name,
            SnapshotType='automated'
        )

        # Sort snapshots by creation time (newest first)
        snapshots = sorted(response['DBSnapshots'], 
            key=lambda s: s['SnapshotCreateTime'], 
            reverse=True
        )

        # Get today's date for naming
        today = datetime.now().strftime("%Y-%m-%d")

        # Copy the newest 'count' snapshots
        for i, snapshot in enumerate(snapshots[:count]):
            if i >= count:
                break

            source_snapshot_id = snapshot['DBSnapshotIdentifier']

            # Create target snapshot name
            target_snapshot_id = f"copy-rds-db-{db_name}-{today}"

            print(f"Copying snapshot {source_snapshot_id} to {target_snapshot_id}")

            try:
            # Create ARN for source snapshot
                source_arn = snapshot['DBSnapshotArn']

            # Copy the snapshot
                dest_rds.copy_db_snapshot(
                    SourceDBSnapshotIdentifier=source_arn,
                    TargetDBSnapshotIdentifier=target_snapshot_id,
                    KmsKeyId="arn:aws:kms:us-east-2:123456789012:key/12345678-asdf-ghjk-zxcv-123456789012",
                    SourceRegion=SOURCE_REGION
                )
                print(f"Successfully initiated copy of {source_snapshot_id} to {target_snapshot_id}")
            except Exception as e:
                print(f"Error copying snapshot {source_snapshot_id}: {str(e)}")

The whole code looks like this:

import boto3
import os
from datetime import datetime

# Define regions
# SOURCE_REGION = "us-west-1"  # N. California
SOURCE_REGION = "us-west-2"  # Oregon
DEST_REGION = "us-east-2"    # Oregon
NUM_SNAPSHOTS = 2            # Number of snapshots to process

def lambda_handler(event, context):
# Create RDS clients for both regions
    source_rds = boto3.client('rds', region_name=SOURCE_REGION)
    dest_rds = boto3.client('rds', region_name=DEST_REGION)

# Step 1: Delete oldest manual snapshots in destination region
    delete_oldest_snapshots(dest_rds, NUM_SNAPSHOTS)

# Step 2: Copy latest system snapshots from source to destination
    copy_latest_snapshots(source_rds, dest_rds, NUM_SNAPSHOTS)

    return {
        'statusCode': 200,
        'body': f'Successfully processed {NUM_SNAPSHOTS} snapshots'
    }

def delete_oldest_snapshots(dest_rds, count):

    db_names = [
        "demo", "drtest20240314"
    ]

# Get all manual snapshots in destination region
    for db_name in db_names:
        response = dest_rds.describe_db_snapshots(
            DBInstanceIdentifier=db_name,
            SnapshotType='manual'
        )

    # Sort snapshots by creation time (oldest first)
        snapshots = sorted(response['DBSnapshots'], key=lambda s: s['SnapshotCreateTime'])

    # Delete the oldest 'count' snapshots
        for i, snapshot in enumerate(snapshots):
            if i >= count:
                break

            snapshot_id = snapshot['DBSnapshotIdentifier']
            print(f"Deleting snapshot: {snapshot_id}")

            try:
                dest_rds.delete_db_snapshot(DBSnapshotIdentifier=snapshot_id)
                print(f"Successfully deleted snapshot: {snapshot_id}")
            except Exception as e:
                print(f"Error deleting snapshot {snapshot_id}: {str(e)}")

def copy_latest_snapshots(source_rds, dest_rds, count):
    # Define the database names based on your example
    #  "amgen", "alpine"
    db_names = [
        "demo", "drtest20240314"
    ]

    for db_name in db_names:
    # Get all automated snapshots in source region
        response = source_rds.describe_db_snapshots(
            DBInstanceIdentifier=db_name,
            SnapshotType='automated'
        )

        # Sort snapshots by creation time (newest first)
        snapshots = sorted(response['DBSnapshots'], 
            key=lambda s: s['SnapshotCreateTime'], 
            reverse=True
        )

        # Get today's date for naming
        today = datetime.now().strftime("%Y-%m-%d")

        # Copy the newest 'count' snapshots
        for i, snapshot in enumerate(snapshots[:count]):
            if i >= count:
                break

            source_snapshot_id = snapshot['DBSnapshotIdentifier']

            # Create target snapshot name
            target_snapshot_id = f"copy-rds-db-{db_name}-{today}"

            print(f"Copying snapshot {source_snapshot_id} to {target_snapshot_id}")

            try:
            # Create ARN for source snapshot
                source_arn = snapshot['DBSnapshotArn']

            # Copy the snapshot
                dest_rds.copy_db_snapshot(
                    SourceDBSnapshotIdentifier=source_arn,
                    TargetDBSnapshotIdentifier=target_snapshot_id,
                    KmsKeyId="arn:aws:kms:us-east-2:123456789012:key/12345678-asdf-ghjk-zxcv-123456789012",
                    SourceRegion=SOURCE_REGION
                )
                print(f"Successfully initiated copy of {source_snapshot_id} to {target_snapshot_id}")
            except Exception as e:
                print(f"Error copying snapshot {source_snapshot_id}: {str(e)}")