Created
September 29, 2019 03:50
-
-
Save kimbo/f604b5bd93fe211c494c4da5cabd3aac to your computer and use it in GitHub Desktop.
simple s3 file upload script
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
import argparse | |
import os | |
import sys | |
import threading | |
import boto3 | |
from botocore.exceptions import ClientError | |
def get_buckets(): | |
return [b.name for b in boto3.resource('s3').buckets.all()] | |
def upload_file(file_name, bucket, object_name=None, extra_args=None): | |
"""Upload a file to an S3 bucket | |
:param file_name: File to upload | |
:param bucket: Bucket to upload to | |
:param object_name: S3 object name. If not specified then file_name is used | |
:return: True if file was uploaded, else False | |
""" | |
# If S3 object_name was not specified, use file_name | |
if object_name is None: | |
object_name = file_name | |
# Upload the file | |
s3_client = boto3.client('s3') | |
s3_client.upload_file(file_name, bucket, object_name, | |
ExtraArgs=extra_args) | |
location = s3_client.get_bucket_location(Bucket=bucket)['LocationConstraint'] | |
return f'https://{bucket}.s3-{location}.amazonaws.com/{object_name}' | |
def main(): | |
buckets = get_buckets() | |
parser = argparse.ArgumentParser() | |
parser.add_argument('file', help='File including latest status update for the week', nargs='?') | |
parser.add_argument('bucket', help='S3 bucket to upload %(file) to', nargs='?', choices=buckets) | |
parser.add_argument('--folder', help='Folder to put the file into. Example: "my-files"') | |
parser.add_argument('--public-read', help='Make file readable by the public (default is not readable by the public', action='store_true') | |
args = parser.parse_args() | |
if args.file is None: | |
args.file = input('File to upload: ') | |
if args.bucket is None: | |
print('Choose a bucket to upload your file to', file=sys.stderr) | |
print('Buckets available:', file=sys.stderr) | |
print(file=sys.stderr) | |
for i, b in enumerate(buckets): | |
print(f'({i+1}): {b}', file=sys.stderr) | |
try: | |
print(file=sys.stderr) | |
idx = int(input(f'Enter your choice (1-{len(buckets)}): ')) | |
if idx > len(buckets): | |
raise ValueError | |
except (ValueError, EOFError): | |
print('Invalid bucket', file=sys.stderr) | |
exit(1) | |
else: | |
args.bucket = buckets[idx-1] | |
object_name = None | |
if args.folder is not None: | |
object_name = f'{args.folder}/{args.file}' | |
extra_args = {} | |
if args.public_read is True: | |
extra_args = {'ACL': 'public-read'} | |
try: | |
url = upload_file(args.file, args.bucket, object_name, extra_args) | |
print(f'Success!!\nURL: {url}') | |
except ClientError as e: | |
print(f'FAILURE! Failed to upload "{args.file}" to bucket "{args.bucket}"', file=sys.stderr) | |
raise e | |
if __name__ == '__main__': | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment