presigned_upload_part async ¶

async def presigned_upload_part(
    self,
    bucket: "capo_s3.types.bucket_name.BucketName",
    key: "capo_s3.types.object_key.ObjectKey",
    part_number: "capo_s3.types.part_number.PartNumber",
    upload_id: "capo_s3.types.multipart_upload_id.MultipartUploadId",
    expire_in: int = 3600,
    *,
    config_overrides: Optional[AsyncS3ClientConfig] = None,
    content_length: Optional["capo_s3.types.content_length.ContentLength"] = None,
    content_md5: Optional["capo_s3.types.content_md5.ContentMD5"] = None,
    checksum_algorithm: Optional[
        "capo_s3.types.checksum_algorithm.ChecksumAlgorithm"
    ] = None,
    checksum_crc32: Optional["capo_s3.types.checksum_crc32.ChecksumCRC32"] = None,
    checksum_crc32_c: Optional[
        "capo_s3.types.checksum_crc32_c.ChecksumCRC32C"
    ] = None,
    checksum_crc64_nvme: Optional[
        "capo_s3.types.checksum_crc64_nvme.ChecksumCRC64NVME"
    ] = None,
    checksum_sha1: Optional["capo_s3.types.checksum_sha1.ChecksumSHA1"] = None,
    checksum_sha256: Optional[
        "capo_s3.types.checksum_sha256.ChecksumSHA256"
    ] = None,
    checksum_sha512: Optional[
        "capo_s3.types.checksum_sha512.ChecksumSHA512"
    ] = None,
    checksum_md5: Optional["capo_s3.types.checksum_md5.ChecksumMD5"] = None,
    checksum_xxhash64: Optional[
        "capo_s3.types.checksum_xxhash64.ChecksumXXHASH64"
    ] = None,
    checksum_xxhash3: Optional[
        "capo_s3.types.checksum_xxhash3.ChecksumXXHASH3"
    ] = None,
    checksum_xxhash128: Optional[
        "capo_s3.types.checksum_xxhash128.ChecksumXXHASH128"
    ] = None,
    sse_customer_algorithm: Optional[
        "capo_s3.types.sse_customer_algorithm.SSECustomerAlgorithm"
    ] = None,
    sse_customer_key: Optional[
        "capo_s3.types.sse_customer_key.SSECustomerKey"
    ] = None,
    sse_customer_key_md5: Optional[
        "capo_s3.types.sse_customer_key_md5.SSECustomerKeyMD5"
    ] = None,
    request_payer: Optional["capo_s3.types.request_payer.RequestPayer"] = None,
    expected_bucket_owner: Optional["capo_s3.types.account_id.AccountId"] = None,
) -> str:
    _, options_ = self.operation_options(config_overrides)
    import base64

    input_: "capo_s3.types.upload_part_request.UploadPartRequest" = {
        "bucket": bucket,
        "key": key,
        "part_number": part_number,
        "upload_id": upload_id,
        "body": ensure_async_iterator(base64.b64decode("")),
    }
    if content_length is not None:
        input_["content_length"] = content_length
    if content_md5 is not None:
        input_["content_md5"] = content_md5
    if checksum_algorithm is not None:
        input_["checksum_algorithm"] = checksum_algorithm
    if checksum_crc32 is not None:
        input_["checksum_crc32"] = checksum_crc32
    if checksum_crc32_c is not None:
        input_["checksum_crc32_c"] = checksum_crc32_c
    if checksum_crc64_nvme is not None:
        input_["checksum_crc64_nvme"] = checksum_crc64_nvme
    if checksum_sha1 is not None:
        input_["checksum_sha1"] = checksum_sha1
    if checksum_sha256 is not None:
        input_["checksum_sha256"] = checksum_sha256
    if checksum_sha512 is not None:
        input_["checksum_sha512"] = checksum_sha512
    if checksum_md5 is not None:
        input_["checksum_md5"] = checksum_md5
    if checksum_xxhash64 is not None:
        input_["checksum_xxhash64"] = checksum_xxhash64
    if checksum_xxhash3 is not None:
        input_["checksum_xxhash3"] = checksum_xxhash3
    if checksum_xxhash128 is not None:
        input_["checksum_xxhash128"] = checksum_xxhash128
    if sse_customer_algorithm is not None:
        input_["sse_customer_algorithm"] = sse_customer_algorithm
    if sse_customer_key is not None:
        input_["sse_customer_key"] = sse_customer_key
    if sse_customer_key_md5 is not None:
        input_["sse_customer_key_md5"] = sse_customer_key_md5
    if request_payer is not None:
        input_["request_payer"] = request_payer
    if expected_bucket_owner is not None:
        input_["expected_bucket_owner"] = expected_bucket_owner

    import capo_s3._operations.amazon_s3.upload_part

    request = capo_s3._operations.amazon_s3.upload_part.build_request(
        options_, input_
    )
    request = strip_checksum_headers(request)
    signer = (request.context or {}).get("signer")
    if not isinstance(signer, SigV4Signer):
        raise RuntimeError("presign requires SigV4 credentials")
    creds = await signer.provider.aresolve_identity()
    ctx: capo_s3._auth._sigv4.SigV4AuthContext = {
        "type": "sig_v4",
        "access_key_id": creds["access_key"],
        "secret_access_key": creds["secret_key"],
        "session_token": creds.get("session_token"),
        "signing_region": signer._auth_scheme["signingRegion"],
        "signing_name": signer._auth_scheme["signingName"],
        "disable_double_encoding": signer._auth_scheme.get(
            "disableDoubleEncoding", False
        ),
        "disable_normalize_path": signer._auth_scheme.get(
            "disableNormalizePath", False
        ),
    }
    return str(presign_sigv4(request, ctx, expires_in=expire_in).url)