Created
December 23, 2020 09:20
-
-
Save lucdew/3e452b992f831d400a655dd7f77cf2ae to your computer and use it in GitHub Desktop.
S3 get all objects versions
This file contains hidden or 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
#!/usr/bin/env python3 | |
import os | |
import boto3 | |
import argparse | |
parser = argparse.ArgumentParser( | |
description="copy all versions of objects into separate files" | |
) | |
parser.add_argument("--bucket", "-b", required=True) | |
parser.add_argument("--profile", "-p") | |
parser.add_argument("--dest", "-d", default=".") | |
parser.add_argument( | |
"objects", nargs="*", default=[], help="objects to get all versions" | |
) | |
args = parse.parse_args() | |
if "AWS_PROFILE" not in os.environ: | |
os.environ["AWS_PROFILE"] = args.profile | |
s3api = boto3.client("s3") | |
for obj in args.objects: | |
paginator = s3api.get_paginator("list_object_versions") | |
versions = [] | |
dest_dir = os.path.join(args.dest, obj) | |
for list_res in paginator.paginate(Bucket=args.bucket, Prefix=obj): | |
versions += [v for v in list_res["Versions"] if v["Key"] == obj] | |
for v in versions: | |
os.makedirs(dest_dir, exist_ok=True) | |
get_res = None | |
try: | |
print(f"Getting key={obj}, version={str(v['LastModified'])}") | |
get_res = s3api.get_object( | |
Bucket=args.bucket, Key=obj, VersionId=v["VersionId"] | |
) | |
except Exception as ex: | |
print(ex) | |
continue | |
sub_dir = str(v["LastModified"]).replace(" ", "-") | |
os.makedirs(os.path.join(dest_dir, sub_dir), exist_ok=True) | |
with open(os.path.join(dest_dir, sub_dir, obj), "wb") as f: | |
f.write(get_res["Body"].read()) | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment