Thursday, March 10, 2022

Python - upload file to S3 using boto3

Get S3 Session


import boto3


def _get_s3_session():
envvars = subprocess.check_output(
['aws-vault', 'exec', 'your-aws-role', '--', 'env'])

aws_access_key_id = ''
aws_secret_access_key = ''
aws_session_token = ''
for envline in envvars.split(b'\n'):
line = envline.decode('utf8')
eqpos = line.find('=')
if eqpos < 4:
continue
k = line[0:eqpos]
v = line[eqpos + 1:]
if k == 'AWS_ACCESS_KEY_ID':
aws_access_key_id = v
if k == 'AWS_SECRET_ACCESS_KEY':
aws_secret_access_key = v
if k == 'AWS_SESSION_TOKEN':
aws_session_token = v

session = boto3.Session(aws_access_key_id, aws_secret_access_key, aws_session_token)
return session

Upload files to S3

class ProgressPercentage(object):
def __init__(self, filename, size=None, prefix_str=''):
self._filename = filename
if size is None:
self._size = float(os.path.getsize(filename))
else:
self._size = size
self._prefix_str = prefix_str
self._seen_so_far = 0
self._lock = threading.Lock()

def __call__(self, bytes_amount):
# To simplify we'll assume this is hooked up
# to a single filename.
with self._lock:
self._seen_so_far += bytes_amount
percentage = (self._seen_so_far / self._size) * 100
sys.stdout.write(
"\r%s %s / %s (%.2f%%)" % (
self._prefix_str,
self._seen_so_far,
self._size,
percentage))
sys.stdout.flush()
def _upload_to_s3(self, **kwargs):
start = time.time()

self._object_name = kwargs['object_name']
self._file_path = kwargs['path']
self._bucket_name = kwargs['bucket_name']

# If S3 object_name was not specified, use file_name
if self._object_name is None:
self._object_name = ntpath.basename(self._file_path )
# Upload the file
session = _get_s3_session()
s3_client = session.client("s3")

try:
# Perform the transfer
s3_client.upload_file(
self._file_path,
self._bucket_name,
self._object_name,
Callback=ProgressPercentage(
self._file_path,
prefix_str='Uploading file {:} -> {:}.{:}: '.format(
os.path.dirname(self._file_path), self._bucket_name, self._object_name)),
)


No comments:

Post a Comment