Add `s3_upload_file` task about prefect-aws HOT 3 CLOSED

import os
from pathlib import Path
import threading
import time

from boto3.s3.transfer import TransferConfig

from prefect.filesystems import S3
from prefect_aws import AwsCredentials


class S3Future(object):
    def __init__(self, filename: Path, final_key: str):
        self.key = final_key
        self._filename = filename
        self._size = filename.stat().st_size
        self._seen_so_far = 0
        self._lock = threading.Lock()
        self._condition = threading.Condition()

    def __call__(self, bytes_amount):
        with self._lock:
            self._seen_so_far += bytes_amount
            if self.done():
                with self._condition():
                    self._condition.notify_all()


    def done(self) -> bool:
        return self._seen_so_far >= self._size

    def wait(self):
        with self._condition():
            self._condition.wait()


def upload_s3(fpath: Path, key: str) -> S3Future:
    aws_credentials = AwsCredentials.load("dataflowops")
    s3_client = aws_credentials.get_boto3_session().client("s3")

    fs: S3 = S3.load("dataflowops")
    bucket, keypath = fs.bucket_path.split("/", 1)
    final_key = "/".join((keypath.rstrip("/"), key.lstrip("/")))
    future = S3Future(fpath, final_key)
    config = TransferConfig(multipart_threshold=1024 * 25, max_concurrency=10,
                            multipart_chunksize=1024 * 25, use_threads=True)
    s3_client.upload_file(str(fpath), bucket, final_key,
                          Config=config,
                          Callback=future,
                          )
    return future