s3upload.py 文件源码

def get_presigned_url(uploader_url, filename, region="us-east-1"):
    """
    Given an uploader URL and a filename, parse the URL and get the presigned URL then use the presigned URL to upload
    the file named filename to S3.

    Parameters:
        uploader_url (str): the uploader URL provided by the AWS engineer
        filename (str): the file to be pushed to S3
        region (str): the S3 endpoint the file should be uploaded to

    Returns:
        presigned_url (str): the presigned URL as a string
    """
    endpoint = "https://30yinsv8k6.execute-api.us-east-1.amazonaws.com/prod/get-signed-url"

    s3uploader_regions = ["us-east-1", "eu-west-1", "ap-northeast-1"]
    if region not in s3uploader_regions:
        region = "us-east-1"

    query_str = urlparse.urlparse(uploader_url).query
    put_dict = urlparse.parse_qs(query_str)

    # If uploader_url is not parsable then maybe it is a shortened URL
    try:
        if not put_dict:
            the_request = urllib_request.Request(uploader_url)
            uploader_url = urllib_urlopen(the_request).geturl()
            query_str = urlparse.urlparse(uploader_url).query
            put_dict = urlparse.parse_qs(query_str)
    except urllib_urlerror:
        pass

    if put_dict:
        # urlparse.parse_qs returns dict values that are single value lists
        # Reassign the values to the first value in the list
        put_dict["accountId"] = put_dict["account-id"][0]
        put_dict["caseId"] = put_dict["case-id"][0]
        put_dict["key"] = put_dict["key"][0]
        put_dict["expiration"] = int(put_dict["expiration"][0])
        put_dict["fileName"] = filename
        put_dict["region"] = region
    else:
        raise S3UploadUrlParsingFailure(uploader_url)

    json_payload = json.dumps(put_dict)

    try:
        response = requests.post(url=endpoint, data=json_payload)
        # If the initial put was successful then proceed with uploading the file using the returned presigned URL
        if response.status_code == 200:
            presigned_url = response.text
            return presigned_url
        else:
            raise S3UploadGetPresignedURLError(response.status_code, uploader_url)
    except requests.exceptions.Timeout:
        raise requests.exceptions.Timeout("ERROR: Connection timed out.")