put_object_annotation async ¶

async def put_object_annotation(
    self,
    bucket: "capo_s3.types.bucket_name.BucketName",
    key: "capo_s3.types.object_key.ObjectKey",
    annotation_name: "capo_s3.types.annotation_name.AnnotationName",
    annotation_payload: Body[AsyncIterator[bytes]] | AsyncIterator[bytes] | bytes,
    *,
    config_overrides: Optional[AsyncS3ClientConfig] = None,
    version_id: Optional["capo_s3.types.object_version_id.ObjectVersionId"] = None,
    object_if_match: Optional["capo_s3.types.object_if_match.ObjectIfMatch"] = 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,
    content_md5: Optional["capo_s3.types.content_md5.ContentMD5"] = None,
    request_payer: Optional["capo_s3.types.request_payer.RequestPayer"] = None,
    expected_bucket_owner: Optional["capo_s3.types.account_id.AccountId"] = None,
) -> "capo_s3.types.put_object_annotation_output.PutObjectAnnotationOutput":
    """<p>Attaches an annotation to an Amazon S3 object. An annotation is a named payload of 1 byte to 1 MiB that you can associate with a specific object or object version. Each object can have up to 1,000 annotations.</p> <p>For annotation naming rules and restrictions, see <a href="https://docs.aws.amazon.com/AmazonS3/latest/userguide/annotations-overview.html">Annotation naming guidelines</a> in the <i>Amazon S3 User Guide</i>.</p> <p>Annotations inherit the encryption of their parent object. For objects without server-side encryption, annotations are encrypted with SSE-S3 (the default for new objects). Objects encrypted with SSE-C cannot have annotations.</p> <p>To use this operation, you must have the <code>s3:PutObjectAnnotation</code> permission. If the bucket has Requester Pays enabled, you must include the <code>x-amz-request-payer</code> header.</p> <note> <p>Annotations are not supported by the following features: S3 Inventory Reports, API Gateway, S3 Storage Lens, Amazon S3 File Gateway, Amazon FSx, S3 on Outposts, and S3 Express One Zone (directory buckets).</p> </note> <p>The following operations are related to <code>PutObjectAnnotation</code>:</p> <ul> <li> <p> <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectAnnotation.html">GetObjectAnnotation</a> </p> </li> <li> <p> <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectAnnotations.html">ListObjectAnnotations</a> </p> </li> <li> <p> <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjectAnnotation.html">DeleteObjectAnnotation</a> </p> </li> </ul>

    Args:
        bucket: <p>The name of the bucket that contains the object.</p>
        key: <p>The object key.</p>
        version_id: <p>The version ID of the object to attach the annotation to.</p>
        annotation_name: <p>The name of the annotation.</p> <p>Length Constraints: Minimum length of 1. Maximum length of 512 bytes.</p>
        annotation_payload: <p>The annotation payload. Must be between 1 byte and 1 MiB in size, and must be valid UTF-8 encoded text. If the payload contains invalid UTF-8 bytes, the request fails with HTTP 415 (Unsupported Media Type). To store binary data, encode the payload using Base64 before uploading.</p>
        object_if_match: <p>If specified, the operation only succeeds if the object's ETag matches the provided value.</p>
        checksum_algorithm: <p>The checksum algorithm to use. Supported values: <code>CRC32</code>, <code>CRC32C</code>, <code>CRC64NVME</code>, <code>SHA1</code>, <code>SHA256</code>, <code>SHA512</code>, <code>MD5</code>, <code>XXHASH64</code>, <code>XXHASH3</code>, <code>XXHASH128</code>.</p>
        checksum_crc32: <p>Base64-encoded CRC32 checksum of the annotation payload.</p>
        checksum_crc32_c: <p>Base64-encoded CRC32C checksum of the annotation payload.</p>
        checksum_crc64_nvme: <p>Base64-encoded CRC64NVME checksum of the annotation payload.</p>
        checksum_sha1: <p>Base64-encoded SHA1 checksum of the annotation payload.</p>
        checksum_sha256: <p>Base64-encoded SHA256 checksum of the annotation payload.</p>
        checksum_sha512: <p>Base64-encoded SHA512 checksum of the annotation payload.</p>
        checksum_md5: <p>Base64-encoded MD5 checksum of the annotation payload.</p>
        checksum_xxhash64: <p>Base64-encoded XXHASH64 checksum of the annotation payload.</p>
        checksum_xxhash3: <p>Base64-encoded XXHASH3 checksum of the annotation payload.</p>
        checksum_xxhash128: <p>Base64-encoded XXHASH128 checksum of the annotation payload.</p>
        content_md5: <p>Base64-encoded MD5 digest of the message.</p>
        expected_bucket_owner: <p>The account ID of the expected bucket owner. If the bucket is owned by a different account, the request fails with an HTTP 403 (Access Denied) error.</p>

    Raises:
        capo_s3.errors.annotation_limit_exceeded.AnnotationLimitExceeded: <p>The request would exceed the maximum number of annotations allowed per object.</p>
        capo_s3.errors.annotation_name_too_long.AnnotationNameTooLong: <p>The annotation name exceeds 512 bytes.</p>
        capo_s3.errors.invalid_annotation_name.InvalidAnnotationName: <p>The annotation name you provided is invalid.</p>
        capo_s3.errors.invalid_request.InvalidRequest: <p>A parameter or header in your request isn't valid. For details, see the description of this API operation.</p>
        capo_s3.errors.no_such_bucket.NoSuchBucket: <p>The specified bucket does not exist.</p>
        capo_s3.errors.no_such_key.NoSuchKey: <p>The specified key does not exist.</p>
        capo_s3.errors.unsupported_media_type.UnsupportedMediaType: <p>The annotation payload is not valid UTF-8 encoded text.</p>
        capo_s3.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_s3.types.put_object_annotation_request.PutObjectAnnotationRequest]",
    ) -> AsyncOperationResponse[
        "capo_s3.types.put_object_annotation_output.PutObjectAnnotationOutput"
    ]:
        import capo_s3._operations.amazon_s3.put_object_annotation

        (
            output,
            http_response,
        ) = await capo_s3._operations.amazon_s3.put_object_annotation.async_put_object_annotation(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_s3.types.put_object_annotation_request.PutObjectAnnotationRequest",
    ) -> dict[str, Any]:
        import capo_s3.types.put_object_annotation_request

        return capo_s3.types.put_object_annotation_request.redact_sensitive(input_)

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("PutObjectAnnotation", redact_input=_redact_input),
    )
    input_: capo_s3.types.put_object_annotation_request.PutObjectAnnotationRequest = {
        "bucket": bucket,
        "key": key,
        "annotation_name": annotation_name,
        "annotation_payload": ensure_async_iterator(annotation_payload),
    }
    if version_id is not None:
        input_["version_id"] = version_id
    if object_if_match is not None:
        input_["object_if_match"] = object_if_match
    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 content_md5 is not None:
        input_["content_md5"] = content_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

    async with aclosing_bodies(input_):
        response = await aexecute_pipeline(
            AsyncOperationRequest(input=input_, options=options_),
            handler=_handler,
            interceptors=list(interceptors_),
        )
        await response.response.aclose()
        return response.output