Download a huge file to the user's computer in chunks from S3

flask_stream_s3.py

import boto3
from boto3.s3.transfer import S3Transfer

from flask import Flask, Response

app = Flask(__name__)


class S3Storage:
    def __init__(self, bucket_name, pictures_path="", video_path=""):
        self.bucket_name = bucket_name
        self.s3 = boto3.client(
            "s3",
            aws_access_key_id="",
            aws_secret_access_key="",
        )
        self.s3_bucket = S3Transfer(self.s3)
        self.baby_pictures_path = pictures_path
        self.guardian_video_path = video_path
        self.MAX_SINGLE_DOWNLOAD_FILESIZE = 1_000_000
        self.MAX_STREAM_SIZE = 1_000_000

    def upload(self, src_path, dst_path):
        self.s3_bucket.upload_file(
            src_path,
            self.bucket_name,
            dst_path,
            extra_args={"ServerSideEncryption": "AES256"},
        )

    def download(self, src_path, dst_path):
        self.s3_bucket.download_file(self.bucket_name, src_path, dst_path)

    def __get_total_bytes(self, bucket_name, file):
        result = self.s3.list_objects(Bucket=bucket_name)
        for item in result["Contents"]:
            if item["Key"] == file:
                return item["Size"]

    def _get_object_range(self, bucket_name, file, total_bytes):
        offset = 0
        stream_size = self.MAX_STREAM_SIZE
        while total_bytes > 0:
            end = offset + stream_size - 1 if total_bytes > stream_size else ""
            total_bytes -= stream_size
            byte_range = "bytes={offset}-{end}".format(offset=offset, end=end)
            offset = end + 1 if not isinstance(end, str) else None
            yield self.s3.get_object(Bucket=bucket_name, Key=file, Range=byte_range)[
                "Body"
            ].read()

    def get_object(self, bucket_name, file):
        total_bytes = self.__get_total_bytes(bucket_name, file)
        if total_bytes > self.MAX_SINGLE_DOWNLOAD_FILESIZE:
            return self._get_object_range(bucket_name, file, total_bytes)
        return self.s3.get_object(Bucket=bucket_name, Key=file)["Body"].read()


bucket_name = ""
file = ""

s3_storage = S3Storage(bucket_name)


@app.route("/get_file", methods=["GET"])
def get_file():

    return Response(
        s3_storage.get_object(bucket_name, file),
        # TODO: Auto detect MIME
        mimetype="video/avi",
        headers={"Content-Disposition": "attachment;filename=" + file},
    )


app.run(debug=True, port=8800)