[Python] Upload large files (multipart) with md5 verification, using same code for both AWS S3...

I'm hoping to upload (potentially) large files (up to 5 Gig) to both Amazon S3 and Alibaba OSS, with verification - preferably MD5 verification.

My code currently looks like:

#!/usr/bin/env python3.10

"""
This toy program is an exploration of how to do verified, multipart file uploads to AWS S3 and Alibaba OSS.

It's a modified version of https://stackoverflow.com/questions/58921396/boto3-multipart-upload-and-md5-checking
"""

import os
import sys
import hashlib

import boto3
from botocore.exceptions import ClientError
from botocore.client import Config
from boto3.s3.transfer import TransferConfig


chunk_size = 2**23


# This function is a re-worked function taken from here: https://stackoverflow.com/questions/43794838/multipart-upload-to-s3-with-hash-verification
# Credits to user: https://stackoverflow.com/users/518169/hyperknot
def calculate_s3_etag(file_path, chunk_size=chunk_size):
"""Calculate an S3/OSS etag for the file."""
chunk_md5s = []

with open(file_path, "rb") as fp:
while True:
data = fp.read(chunk_size)

if not data:
break

chunk_md5s.append(hashlib.md5(data))

num_hashes = len(chunk_md5s)

if not num_hashes:
# do whatever you want to do here
raise ValueError

if num_hashes == 1:
return f"{chunk_md5s[0].hexdigest()}"

digest_byte_string = b"".join(m.digest() for m in chunk_md5s)
digests_md5 = hashlib.md5(digest_byte_string)

return f"{digests_md5.hexdigest().lower()}-{num_hashes}"


def s3_md5sum(bucket_name, resource_name, client):
"""Calculate an MD5 hash for the file."""
try:
et = client.head_object(Bucket=bucket_name, Key=resource_name)["ETag"]
assert et[0] == '"' and et[-1] == '"'
return et[1:-1].lower()
except ClientError:
# do whatever you want to do here
raise ClientError


def upload_one_file(
*,
filename,
endpoint_url,
aws_credentials,
aws_region,
bucket,
addressing_style=None,
):
"""Upload one file to S3. The file can be up to 5 gigabytes in size, and will be verified on upload."""
config = {}
config["region_name"] = aws_region
if addressing_style:
config["s3"] = {"addressing_style": addressing_style}
kwargs = {"config": Config(**config), **aws_credentials}
if endpoint_url:
kwargs["endpoint_url"] = endpoint_url
client = boto3.client("s3", **kwargs)
transfer_config = TransferConfig(multipart_chunksize=chunk_size)

client.upload_file(filename, bucket, filename, Config=transfer_config)

tag = calculate_s3_etag(filename)
result = s3_md5sum(bucket, filename, client)

return (tag, result)


class AuthorizationData:
"""Hold things about what S3/OSS to use."""

def __init__(
self, *, name, bucket, aws_region, aws_access_key_id, aws_secret_access_key, endpoint_url=None, addressing_style=None
):
"""Initialize."""
self.name = name
self.endpoint_url = endpoint_url
self.bucket = bucket
self.aws_region = aws_region
# self.aws_credentials = aws_credentials
self.aws_credentials = {
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
}
self.addressing_style = addressing_style


def create_test_inputs():
"""Create test files."""
for filename, blocksize, total_size in (
("dastromberg-urandom-1k", 2**10, 2**10),
("dastromberg-urandom-16M", 2**20, 2**24),
("dastromberg-urandom-256M", 2**20, 2**28),
("dastromberg-urandom-1G", 2**20, 2**30),
("dastromberg-urandom-5G", 2**20, 5 * 2**30),
):
if not os.path.isfile(filename):
length_written = 0
print("Creating test input %s" % filename)
with (
open("/dev/urandom", "rb") as infile,
open(filename, "wb") as outfile,
):
while length_written < total_size:
block = infile.read(blocksize)
outfile.write(block)
len_block = len(block)
assert blocksize == len_block
length_written += len_block

assert length_written == total_size

assert os.path.getsize(filename) == total_size, "Final length does not match. Perhaps remove %s and try again?" % filename


def main():
"""Test some AWS S3- and Alibaba OSS-use."""
# This works.
aws = AuthorizationData(
name="AWS",
bucket="aaa",
aws_region="us-west-2",
aws_access_key_id="bbb",
aws_secret_access_key=os.environ["aws_secret_key"],
)

alibaba = AuthorizationData(
name="Alibaba",
# endpoint_url="https://blz-p-cdn-gamepublishing.oss-cn-hangzhou.aliyuncs.com/",
endpoint_url="https://oss-cn-hangzhou.aliyuncs.com",
bucket="ccc",
aws_region="cn-hangzhou",
aws_access_key_id="ddd",
aws_secret_access_key=os.environ["alibaba_secret_key"],
addressing_style="virtual",
)

all_good = True

create_test_inputs()

for auth in (
aws,
alibaba,
):
for filename in (
"dastromberg-urandom-1k",
"dastromberg-urandom-16M",
"dastromberg-urandom-256M",
# "dastromberg-urandom-1G",
# "dastromberg-urandom-5G",
):
print(f"Checking {auth.name}: {filename}")
(tag, result) = upload_one_file(
filename=filename,
endpoint_url=auth.endpoint_url,
aws_credentials=auth.aws_credentials,
aws_region=auth.aws_region,
bucket=auth.bucket,
addressing_style=auth.addressing_style,
)

if tag == result:
print("Verification succeeded: %s, %s" % (tag, result))
else:
all_good = False
print("Verification failed: %s, %s" % (tag, result), file=sys.stderr)

if all_good:
print("All tests passed")
else:
print("One or more tests failed", file=sys.stderr)
sys.exit(1)


if __name__ == "__main__":
main()


Running it, AWS S3 verifies for all test inputs, but Alibaba only verifies for the 1 kilobyte test - the 16 megabyte test fails, as do the larger tests.

In https://www.alibabacloud.com/help/e...s-as-oss-md5-hashes-to-check-data-consistency Alibaba says that they don't recommend using MD5 for verification.

In https://www.alibabacloud.com/help/en/oss/use-cases/check-data-transmission-integrity-by-using-crc-64 they seem to relatively thoroughly explore using x-oss-hash-crc64ecma. I'd rather use MD5 (with boto3), but can use crc64ecma if necessary (hopefully with boto3 still).

Has anyone explored this?

Any suggestions?

Continue reading...