get_object_annotation async ¶

@asynccontextmanager
async def get_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",
    *,
    config_overrides: Optional[AsyncS3ClientConfig] = None,
    version_id: Optional["capo_s3.types.object_version_id.ObjectVersionId"] = None,
    request_payer: Optional["capo_s3.types.request_payer.RequestPayer"] = None,
    expected_bucket_owner: Optional["capo_s3.types.account_id.AccountId"] = None,
    checksum_mode: Optional["capo_s3.types.checksum_mode.ChecksumMode"] = None,
) -> "AsyncGenerator[capo_s3.types.get_object_annotation_output.GetObjectAnnotationOutput]":
    """<p>Retrieves an annotation from an Amazon S3 object. To use this operation, you must have the <code>s3:GetObjectAnnotation</code> permission.</p> <p>If checksum mode is enabled via the <code>x-amz-checksum-mode</code> header, Amazon S3 returns the stored checksum in the response headers for client-side validation.</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>GetObjectAnnotation</code>:</p> <ul> <li> <p> <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectAnnotation.html">PutObjectAnnotation</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>
        annotation_name: <p>The name of the annotation to retrieve.</p> <p>Length Constraints: Minimum length of 1. Maximum length of 512 bytes.</p>
        version_id: <p>The version ID of the object.</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>
        checksum_mode: <p>Set to <code>ENABLED</code> to validate the checksum of the annotation payload on retrieval.</p>

    Raises:
        capo_s3.errors.no_such_annotation.NoSuchAnnotation: <p>The specified annotation does not exist on this object.</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.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_s3.types.get_object_annotation_request.GetObjectAnnotationRequest]",
    ) -> AsyncOperationResponse[
        "capo_s3.types.get_object_annotation_output.GetObjectAnnotationOutput"
    ]:
        import capo_s3._operations.amazon_s3.get_object_annotation

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

    def _redact_output(
        output: "capo_s3.types.get_object_annotation_output.GetObjectAnnotationOutput",
    ) -> dict[str, Any]:
        import capo_s3.types.get_object_annotation_output

        return capo_s3.types.get_object_annotation_output.redact_sensitive(output)

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("GetObjectAnnotation", redact_output=_redact_output),
    )
    input_: capo_s3.types.get_object_annotation_request.GetObjectAnnotationRequest = {
        "bucket": bucket,
        "key": key,
        "annotation_name": annotation_name,
    }
    if version_id is not None:
        input_["version_id"] = version_id
    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
    if checksum_mode is not None:
        input_["checksum_mode"] = checksum_mode

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    try:
        yield response.output
    finally:
        await response.response.aclose()