Skip to content

Instantly share code, notes, and snippets.

@kimbo
Created September 29, 2019 03:50
Show Gist options
  • Save kimbo/f604b5bd93fe211c494c4da5cabd3aac to your computer and use it in GitHub Desktop.
Save kimbo/f604b5bd93fe211c494c4da5cabd3aac to your computer and use it in GitHub Desktop.
simple s3 file upload script
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