Skip to content

Instantly share code, notes, and snippets.

@JBirdVegas
Created September 14, 2018 05:22
Show Gist options
  • Save JBirdVegas/89b4a73d064ca6f9512855f630fbaa4b to your computer and use it in GitHub Desktop.
Save JBirdVegas/89b4a73d064ca6f9512855f630fbaa4b to your computer and use it in GitHub Desktop.
Bulk undelete of versioned s3 objects
#!/usr/bin/env python
import json
from datetime import datetime
from boto3 import Session
BUCKET_NAME = "my-sweet-bucket"
prefix = 'my-sweet-dir'
session = Session(region_name='us-east-1', profile_name='my.sweet.profile.name')
bucket = session.client('s3')
MAX_KEYS = 10000
def get_bucket_versions(version_id, key):
return bucket.list_object_versions(Bucket=BUCKET_NAME,
MaxKeys=MAX_KEYS,
Prefix=prefix,
VersionIdMarker=version_id,
KeyMarker=key)
class DateTimeEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, datetime):
return o.isoformat()
return json.JSONEncoder.default(self, o)
def objects_to_restore(versions):
return [
{
'VersionId': marker['VersionId'],
'Key': marker['Key']
} for marker in versions.get('DeleteMarkers') if marker['IsLatest']
]
def restore_s3_objects(version_markers, count):
markers_to_delete = objects_to_restore(version_markers)
print(f"Will restore {len(markers_to_delete)} items during request number: {count}")
if not markers_to_delete:
return 0
bucket.delete_objects(Bucket=BUCKET_NAME, Delete={'Objects': markers_to_delete})
return len(markers_to_delete)
obj_list = bucket.list_object_versions(Bucket=BUCKET_NAME,
MaxKeys=MAX_KEYS,
Prefix=prefix)
_next_version_id = obj_list.get('NextVersionIdMarker')
_next_key_marker = obj_list.get('NextKeyMarker')
counter = 1
total_restored = restore_s3_objects(obj_list, counter)
while _next_version_id and _next_key_marker:
counter += 1
another_list_of_versions = get_bucket_versions(_next_version_id, _next_key_marker)
_next_version_id = another_list_of_versions.get('NextVersionIdMarker')
_next_key_marker = another_list_of_versions.get('NextKeyMarker')
total_restored += restore_s3_objects(another_list_of_versions, counter)
print(f"Total Restored: {total_restored}")
@auskoki
Copy link

auskoki commented Apr 27, 2022

ciao and thanks for this really usefull script.
Just a question : i' m really basic with python but i'm trying anyway without success until now to add another filter just to delete all the "delete marker" placeholder filtered by datetime.
I strted to work with the json field "LastModified" but i 'm still far away from the solution
may be can you help me ?

ciao and regards
Germano

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment