list_async_invokes

Operation

list_async_invokes async

list_async_invokes(input: ListAsyncInvokesInput, plugins: list[Plugin] | None = None) -> ListAsyncInvokesOutput

Lists asynchronous invocations.

Parameters:

Name	Type	Description	Default
input	ListAsyncInvokesInput	An instance of ListAsyncInvokesInput.	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
ListAsyncInvokesOutput	An instance of ListAsyncInvokesOutput.

Source code in src/aws_sdk_bedrock_runtime/client.py

async def list_async_invokes(
    self, input: ListAsyncInvokesInput, plugins: list[Plugin] | None = None
) -> ListAsyncInvokesOutput:
    """Lists asynchronous invocations.

    Args:
        input:
            An instance of `ListAsyncInvokesInput`.
        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 `ListAsyncInvokesOutput`.
    """
    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=LIST_ASYNC_INVOKES,
        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

ListAsyncInvokesInput dataclass

Dataclass for ListAsyncInvokesInput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

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

    submit_time_after: datetime | None = None
    """Include invocations submitted after this time."""

    submit_time_before: datetime | None = None
    """Include invocations submitted before this time."""

    status_equals: str | None = None
    """Filter invocations by status."""

    max_results: int | None = None
    """The maximum number of invocations to return in one page of results."""

    next_token: str | None = None
    """Specify the pagination token from a previous request to retrieve the
    next page of results.
    """

    sort_by: str = "SubmissionTime"
    """How to sort the response."""

    sort_order: str = "Descending"
    """The sorting order for the response."""

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

    def serialize_members(self, serializer: ShapeSerializer):
        if self.submit_time_after is not None:
            serializer.write_timestamp(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["submitTimeAfter"],
                self.submit_time_after,
            )

        if self.submit_time_before is not None:
            serializer.write_timestamp(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["submitTimeBefore"],
                self.submit_time_before,
            )

        if self.status_equals is not None:
            serializer.write_string(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["statusEquals"],
                self.status_equals,
            )

        if self.max_results is not None:
            serializer.write_integer(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["maxResults"], self.max_results
            )

        if self.next_token is not None:
            serializer.write_string(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["nextToken"], self.next_token
            )

        if self.sort_by is not None:
            serializer.write_string(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["sortBy"], self.sort_by
            )

        if self.sort_order is not None:
            serializer.write_string(
                _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["sortOrder"], self.sort_order
            )

    @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["submit_time_after"] = de.read_timestamp(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["submitTimeAfter"]
                    )

                case 1:
                    kwargs["submit_time_before"] = de.read_timestamp(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["submitTimeBefore"]
                    )

                case 2:
                    kwargs["status_equals"] = de.read_string(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["statusEquals"]
                    )

                case 3:
                    kwargs["max_results"] = de.read_integer(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["maxResults"]
                    )

                case 4:
                    kwargs["next_token"] = de.read_string(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["nextToken"]
                    )

                case 5:
                    kwargs["sort_by"] = de.read_string(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["sortBy"]
                    )

                case 6:
                    kwargs["sort_order"] = de.read_string(
                        _SCHEMA_LIST_ASYNC_INVOKES_INPUT.members["sortOrder"]
                    )

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

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

Attributes

max_results class-attribute instance-attribute

max_results: int | None = None

The maximum number of invocations to return in one page of results.

next_token class-attribute instance-attribute

next_token: str | None = None

Specify the pagination token from a previous request to retrieve the next page of results.

sort_by class-attribute instance-attribute

sort_by: str = 'SubmissionTime'

How to sort the response.

sort_order class-attribute instance-attribute

sort_order: str = 'Descending'

The sorting order for the response.

status_equals class-attribute instance-attribute

status_equals: str | None = None

Filter invocations by status.

submit_time_after class-attribute instance-attribute

submit_time_after: datetime | None = None

Include invocations submitted after this time.

submit_time_before class-attribute instance-attribute

submit_time_before: datetime | None = None

Include invocations submitted before this time.

Output

ListAsyncInvokesOutput dataclass

Dataclass for ListAsyncInvokesOutput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

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

    next_token: str | None = None
    """Specify the pagination token from a previous request to retrieve the
    next page of results.
    """

    async_invoke_summaries: list[AsyncInvokeSummary] | None = None
    """A list of invocation summaries."""

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

    def serialize_members(self, serializer: ShapeSerializer):
        if self.next_token is not None:
            serializer.write_string(
                _SCHEMA_LIST_ASYNC_INVOKES_OUTPUT.members["nextToken"], self.next_token
            )

        if self.async_invoke_summaries is not None:
            _serialize_async_invoke_summaries(
                serializer,
                _SCHEMA_LIST_ASYNC_INVOKES_OUTPUT.members["asyncInvokeSummaries"],
                self.async_invoke_summaries,
            )

    @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["next_token"] = de.read_string(
                        _SCHEMA_LIST_ASYNC_INVOKES_OUTPUT.members["nextToken"]
                    )

                case 1:
                    kwargs["async_invoke_summaries"] = (
                        _deserialize_async_invoke_summaries(
                            de,
                            _SCHEMA_LIST_ASYNC_INVOKES_OUTPUT.members[
                                "asyncInvokeSummaries"
                            ],
                        )
                    )

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

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

Attributes

async_invoke_summaries class-attribute instance-attribute

async_invoke_summaries: list[AsyncInvokeSummary] | None = None

A list of invocation summaries.

next_token class-attribute instance-attribute

next_token: str | None = None

Specify the pagination token from a previous request to retrieve the next page of results.

Errors

• AccessDeniedException

• InternalServerException

• ThrottlingException

• ValidationException