Last active
February 23, 2023 20:17
-
-
Save CodaBool/a057a848d5333f21feefee8ffc2a6f02 to your computer and use it in GitHub Desktop.
restores s3 files by deleteing key delete marker
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from datetime import datetime, timezone | |
import boto3 | |
# ----------------------------------- | |
# Enter these values here: | |
thebucket = "BUCKET_NAME_HERE" | |
region_name = "us-east-1" | |
folder_in_bucket = "" # aka key or prefix | |
# ------------------------------------ | |
# can optionally write these values if they are not in environment | |
# access_key = "<access_key>" | |
# secret_key = "<secret_key>" | |
# s3 = boto3.resource("s3", region_name=region_name, aws_access_key_id=access_key, aws_secret_access_key=secret_key) | |
# s3client = boto3.client("s3", aws_access_key_id=access_key, aws_secret_access_key=secret_key) | |
s3 = boto3.resource("s3", region_name=region_name) | |
s3client = boto3.client("s3") | |
page_size = 100000 | |
paginator = s3client.get_paginator("list_object_versions") | |
pageresponse = paginator.paginate( | |
Bucket=thebucket, PaginationConfig={"MaxItems": page_size} | |
) | |
deleted_at = datetime(2019, 10, 22, 20, 0, 0, tzinfo=timezone.utc) | |
def time_ago(time=False): | |
now = datetime.now(timezone.utc) | |
if type(time) is int: | |
diff = now - datetime.fromtimestamp(time) | |
elif isinstance(time,datetime): | |
diff = now - time | |
elif not time: | |
diff = now - now | |
else: | |
raise ValueError('invalid date %s of type %s' % (time, type(time))) | |
second_diff = diff.seconds | |
day_diff = diff.days | |
if day_diff < 0: | |
return '' | |
if day_diff == 0: | |
if second_diff < 10: | |
return "just now" | |
if second_diff < 60: | |
return str(round(second_diff)) + " seconds ago" | |
if second_diff < 120: | |
return "a minute ago" | |
if second_diff < 3600: | |
return str( round(second_diff / 60 )) + " minutes ago" | |
if second_diff < 7200: | |
return "an hour ago" | |
if second_diff < 86400: | |
return str( round(second_diff / 3600 )) + " hours ago" | |
if day_diff == 1: | |
return "Yesterday" | |
if day_diff < 7: | |
return str(round(day_diff)) + " days ago" | |
if day_diff < 31: | |
return str(round(day_diff/7)) + " weeks ago" | |
if day_diff < 365: | |
return str(round(day_diff/30)) + " months ago" | |
return str(round(day_diff/365)) + " years ago" | |
def restore_all(pages): | |
# iter over the pages from the paginator | |
for page in pages: | |
# Find if there are any delmarkers | |
if "DeleteMarkers" in page.keys(): | |
for each_delmarker in page["DeleteMarkers"]: | |
print(f"\n📄 {each_delmarker['Key']}") | |
if each_delmarker["IsLatest"] is True and each_delmarker["LastModified"] > deleted_at: | |
print(f" This was deleted {time_ago(each_delmarker['LastModified'])}") | |
restore(each_delmarker) | |
def restore(delete_marker): | |
answer = input(" Would you like to restore this file? [y/N] ") or "N" | |
if answer == "y" or answer == "Y": | |
print(" 🌱 Restoring...") | |
file_object_version = s3.ObjectVersion(thebucket, delete_marker["Key"], delete_marker["VersionId"]) | |
file_object_version.delete() | |
else: | |
print(" 👟 Skipping...") | |
if __name__ == "__main__": | |
print(f"Restoring files deleted from {time_ago(deleted_at)} in 🪣 {thebucket}/{folder_in_bucket}") | |
restore_all(pageresponse) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment