Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import boto3
import click
import os
class DownLoadS3Objects(object):
""" copy s3 object """
def __init__(self, profile, bucket_name, destination, prefix):
self.session = boto3.Session(profile_name=profile)
self.s3 = self.session.resource("s3")
self.bucket_name = bucket_name
self.bucket = self.s3.Bucket(bucket_name)
self.dst = destination
self.prefix = prefix
def get_object_list(self):
return self.bucket.objects.filter(Prefix=self.prefix)
def download_object(self, s3obj):
file_path = os.path.join(self.dst, s3obj.key.split("/")[-1])
self.s3.Object(s3obj.bucket_name, s3obj.key).download_file(file_path)
print("downloaded {}".format(file_path))
def main(self):
for o in self.get_object_list():
self.download_object(o)
@click.command()
@click.option("--profile", "-p", default="default")
@click.option("--bucket-name", "-b")
@click.option("--destination", "-d", default="./")
@click.option("--prefix", "-P")
def cmd(profile, bucket_name, destination, prefix):
d = DownLoadS3Objects(profile, bucket_name, destination, prefix)
d.main()
def main():
cmd()
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.