get_async_invoke

Operation

get_async_invoke async

get_async_invoke(input: GetAsyncInvokeInput, plugins: list[Plugin] | None = None) -> GetAsyncInvokeOutput

Retrieve information about an asynchronous invocation.

Parameters:

Name	Type	Description	Default
input	GetAsyncInvokeInput	An instance of GetAsyncInvokeInput.	required
plugins	list[Plugin] | None	A list of callables that modify the configuration dynamically. Changes made by these plugins only apply for the duration of the operation execution and will not affect any other operation invocations.	None

Returns:

Type	Description
GetAsyncInvokeOutput	An instance of GetAsyncInvokeOutput.

Source code in src/aws_sdk_bedrock_runtime/client.py

async def get_async_invoke(
    self, input: GetAsyncInvokeInput, plugins: list[Plugin] | None = None
) -> GetAsyncInvokeOutput:
    """Retrieve information about an asynchronous invocation.

    Args:
        input:
            An instance of `GetAsyncInvokeInput`.
        plugins:
            A list of callables that modify the configuration dynamically.
            Changes made by these plugins only apply for the duration of the
            operation execution and will not affect any other operation
            invocations.

    Returns:
        An instance of `GetAsyncInvokeOutput`.
    """
    operation_plugins: list[Plugin] = []
    if plugins:
        operation_plugins.extend(plugins)
    config = deepcopy(self._config)
    for plugin in operation_plugins:
        plugin(config)
    if config.protocol is None or config.transport is None:
        raise ExpectationNotMetError(
            "protocol and transport MUST be set on the config to make calls."
        )
    pipeline = RequestPipeline(protocol=config.protocol, transport=config.transport)
    call = ClientCall(
        input=input,
        operation=GET_ASYNC_INVOKE,
        context=TypedProperties({"config": config}),
        interceptor=InterceptorChain(config.interceptors),
        auth_scheme_resolver=config.auth_scheme_resolver,
        supported_auth_schemes=config.auth_schemes,
        endpoint_resolver=config.endpoint_resolver,
        retry_strategy=config.retry_strategy,
    )

    return await pipeline(call)

Input

GetAsyncInvokeInput dataclass

Dataclass for GetAsyncInvokeInput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

@dataclass(kw_only=True)
class GetAsyncInvokeInput:
    """Dataclass for GetAsyncInvokeInput structure."""

    invocation_arn: str | None = None
    """The invocation's ARN."""

    def serialize(self, serializer: ShapeSerializer):
        serializer.write_struct(_SCHEMA_GET_ASYNC_INVOKE_INPUT, self)

    def serialize_members(self, serializer: ShapeSerializer):
        if self.invocation_arn is not None:
            serializer.write_string(
                _SCHEMA_GET_ASYNC_INVOKE_INPUT.members["invocationArn"],
                self.invocation_arn,
            )

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        return cls(**cls.deserialize_kwargs(deserializer))

    @classmethod
    def deserialize_kwargs(cls, deserializer: ShapeDeserializer) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}

        def _consumer(schema: Schema, de: ShapeDeserializer) -> None:
            match schema.expect_member_index():
                case 0:
                    kwargs["invocation_arn"] = de.read_string(
                        _SCHEMA_GET_ASYNC_INVOKE_INPUT.members["invocationArn"]
                    )

                case _:
                    logger.debug("Unexpected member schema: %s", schema)

        deserializer.read_struct(_SCHEMA_GET_ASYNC_INVOKE_INPUT, consumer=_consumer)
        return kwargs

Attributes

invocation_arn class-attribute instance-attribute

invocation_arn: str | None = None

The invocation's ARN.

Output

GetAsyncInvokeOutput dataclass

Dataclass for GetAsyncInvokeOutput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

@dataclass(kw_only=True)
class GetAsyncInvokeOutput:
    """Dataclass for GetAsyncInvokeOutput structure."""

    invocation_arn: str
    """The invocation's ARN."""

    model_arn: str
    """The invocation's model ARN."""

    status: str
    """The invocation's status."""

    submit_time: datetime
    """When the invocation request was submitted."""

    output_data_config: AsyncInvokeOutputDataConfig
    """Output data settings."""

    client_request_token: str | None = None
    """The invocation's idempotency token."""

    failure_message: str | None = field(repr=False, default=None)
    """An error message."""

    last_modified_time: datetime | None = None
    """The invocation's last modified time."""

    end_time: datetime | None = None
    """When the invocation ended."""

    def serialize(self, serializer: ShapeSerializer):
        serializer.write_struct(_SCHEMA_GET_ASYNC_INVOKE_OUTPUT, self)

    def serialize_members(self, serializer: ShapeSerializer):
        serializer.write_string(
            _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["invocationArn"],
            self.invocation_arn,
        )
        serializer.write_string(
            _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["modelArn"], self.model_arn
        )
        if self.client_request_token is not None:
            serializer.write_string(
                _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["clientRequestToken"],
                self.client_request_token,
            )

        serializer.write_string(
            _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["status"], self.status
        )
        if self.failure_message is not None:
            serializer.write_string(
                _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["failureMessage"],
                self.failure_message,
            )

        serializer.write_timestamp(
            _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["submitTime"], self.submit_time
        )
        if self.last_modified_time is not None:
            serializer.write_timestamp(
                _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["lastModifiedTime"],
                self.last_modified_time,
            )

        if self.end_time is not None:
            serializer.write_timestamp(
                _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["endTime"], self.end_time
            )

        serializer.write_struct(
            _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["outputDataConfig"],
            self.output_data_config,
        )

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        return cls(**cls.deserialize_kwargs(deserializer))

    @classmethod
    def deserialize_kwargs(cls, deserializer: ShapeDeserializer) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}

        def _consumer(schema: Schema, de: ShapeDeserializer) -> None:
            match schema.expect_member_index():
                case 0:
                    kwargs["invocation_arn"] = de.read_string(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["invocationArn"]
                    )

                case 1:
                    kwargs["model_arn"] = de.read_string(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["modelArn"]
                    )

                case 2:
                    kwargs["client_request_token"] = de.read_string(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["clientRequestToken"]
                    )

                case 3:
                    kwargs["status"] = de.read_string(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["status"]
                    )

                case 4:
                    kwargs["failure_message"] = de.read_string(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["failureMessage"]
                    )

                case 5:
                    kwargs["submit_time"] = de.read_timestamp(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["submitTime"]
                    )

                case 6:
                    kwargs["last_modified_time"] = de.read_timestamp(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["lastModifiedTime"]
                    )

                case 7:
                    kwargs["end_time"] = de.read_timestamp(
                        _SCHEMA_GET_ASYNC_INVOKE_OUTPUT.members["endTime"]
                    )

                case 8:
                    kwargs["output_data_config"] = (
                        _AsyncInvokeOutputDataConfigDeserializer().deserialize(de)
                    )

                case _:
                    logger.debug("Unexpected member schema: %s", schema)

        deserializer.read_struct(_SCHEMA_GET_ASYNC_INVOKE_OUTPUT, consumer=_consumer)
        return kwargs

Attributes

client_request_token class-attribute instance-attribute

client_request_token: str | None = None

The invocation's idempotency token.

end_time class-attribute instance-attribute

end_time: datetime | None = None

When the invocation ended.

failure_message class-attribute instance-attribute

failure_message: str | None = field(repr=False, default=None)

An error message.

invocation_arn instance-attribute

invocation_arn: str

The invocation's ARN.

last_modified_time class-attribute instance-attribute

last_modified_time: datetime | None = None

The invocation's last modified time.

model_arn instance-attribute

model_arn: str

The invocation's model ARN.

output_data_config instance-attribute

output_data_config: AsyncInvokeOutputDataConfig

Output data settings.

status instance-attribute

status: str

The invocation's status.

submit_time instance-attribute

submit_time: datetime

When the invocation request was submitted.

Errors

• AccessDeniedException

• InternalServerException

• ThrottlingException

• ValidationException