presigned_head_object async ¶

async def presigned_head_object(
    self,
    bucket: "capo_s3.types.bucket_name.BucketName",
    key: "capo_s3.types.object_key.ObjectKey",
    expire_in: int = 3600,
    *,
    config_overrides: Optional[AsyncS3ClientConfig] = None,
    if_match: Optional["capo_s3.types.if_match.IfMatch"] = None,
    if_modified_since: Optional[
        "capo_s3.types.if_modified_since.IfModifiedSince"
    ] = None,
    if_none_match: Optional["capo_s3.types.if_none_match.IfNoneMatch"] = None,
    if_unmodified_since: Optional[
        "capo_s3.types.if_unmodified_since.IfUnmodifiedSince"
    ] = None,
    range: Optional["capo_s3.types.range.Range"] = None,
    response_cache_control: Optional[
        "capo_s3.types.response_cache_control.ResponseCacheControl"
    ] = None,
    response_content_disposition: Optional[
        "capo_s3.types.response_content_disposition.ResponseContentDisposition"
    ] = None,
    response_content_encoding: Optional[
        "capo_s3.types.response_content_encoding.ResponseContentEncoding"
    ] = None,
    response_content_language: Optional[
        "capo_s3.types.response_content_language.ResponseContentLanguage"
    ] = None,
    response_content_type: Optional[
        "capo_s3.types.response_content_type.ResponseContentType"
    ] = None,
    response_expires: Optional[
        "capo_s3.types.response_expires.ResponseExpires"
    ] = None,
    version_id: Optional["capo_s3.types.object_version_id.ObjectVersionId"] = 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,
    part_number: Optional["capo_s3.types.part_number.PartNumber"] = None,
    expected_bucket_owner: Optional["capo_s3.types.account_id.AccountId"] = None,
    checksum_mode: Optional["capo_s3.types.checksum_mode.ChecksumMode"] = None,
) -> str:
    _, options_ = self.operation_options(config_overrides)
    input_: "capo_s3.types.head_object_request.HeadObjectRequest" = {
        "bucket": bucket,
        "key": key,
    }
    if if_match is not None:
        input_["if_match"] = if_match
    if if_modified_since is not None:
        input_["if_modified_since"] = if_modified_since
    if if_none_match is not None:
        input_["if_none_match"] = if_none_match
    if if_unmodified_since is not None:
        input_["if_unmodified_since"] = if_unmodified_since
    if range is not None:
        input_["range"] = range
    if response_cache_control is not None:
        input_["response_cache_control"] = response_cache_control
    if response_content_disposition is not None:
        input_["response_content_disposition"] = response_content_disposition
    if response_content_encoding is not None:
        input_["response_content_encoding"] = response_content_encoding
    if response_content_language is not None:
        input_["response_content_language"] = response_content_language
    if response_content_type is not None:
        input_["response_content_type"] = response_content_type
    if response_expires is not None:
        input_["response_expires"] = response_expires
    if version_id is not None:
        input_["version_id"] = version_id
    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 part_number is not None:
        input_["part_number"] = part_number
    if expected_bucket_owner is not None:
        input_["expected_bucket_owner"] = expected_bucket_owner
    if checksum_mode is not None:
        input_["checksum_mode"] = checksum_mode

    import capo_s3._operations.amazon_s3.head_object

    request = capo_s3._operations.amazon_s3.head_object.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)