lib/python3.10/site-packages/aiobotocore/waiter.py

import asyncio

from botocore.docs.docstring import WaiterDocstring

# WaiterModel is required for client.py import
from botocore.exceptions import ClientError
from botocore.utils import get_service_module_name
from botocore.waiter import (
    NormalizedOperationMethod as _NormalizedOperationMethod,
)
from botocore.waiter import WaiterModel  # noqa: F401 lgtm[py/unused-import]
from botocore.waiter import (
    Waiter,
    WaiterError,
    is_valid_waiter_error,
    logger,
    xform_name,
)


def create_waiter_with_client(waiter_name, waiter_model, client):
    """

    :type waiter_name: str
    :param waiter_name: The name of the waiter.  The name should match
        the name (including the casing) of the key name in the waiter
        model file (typically this is CamelCasing).

    :type waiter_model: botocore.waiter.WaiterModel
    :param waiter_model: The model for the waiter configuration.

    :type client: botocore.client.BaseClient
    :param client: The botocore client associated with the service.

    :rtype: botocore.waiter.Waiter
    :return: The waiter object.

    """
    single_waiter_config = waiter_model.get_waiter(waiter_name)
    operation_name = xform_name(single_waiter_config.operation)
    operation_method = NormalizedOperationMethod(
        getattr(client, operation_name)
    )

    # Create a new wait method that will serve as a proxy to the underlying
    # Waiter.wait method. This is needed to attach a docstring to the
    # method.
    async def wait(self, **kwargs):
        return await AIOWaiter.wait(self, **kwargs)

    wait.__doc__ = WaiterDocstring(
        waiter_name=waiter_name,
        event_emitter=client.meta.events,
        service_model=client.meta.service_model,
        service_waiter_model=waiter_model,
        include_signature=False,
    )

    # Rename the waiter class based on the type of waiter.
    waiter_class_name = str(
        '%s.Waiter.%s'
        % (get_service_module_name(client.meta.service_model), waiter_name)
    )

    # Create the new waiter class
    documented_waiter_cls = type(waiter_class_name, (Waiter,), {'wait': wait})

    # Return an instance of the new waiter class.
    return documented_waiter_cls(
        waiter_name, single_waiter_config, operation_method
    )


class NormalizedOperationMethod(_NormalizedOperationMethod):
    async def __call__(self, **kwargs):
        try:
            return await self._client_method(**kwargs)
        except ClientError as e:
            return e.response


class AIOWaiter(Waiter):
    async def wait(self, **kwargs):
        acceptors = list(self.config.acceptors)
        current_state = 'waiting'
        # pop the invocation specific config
        config = kwargs.pop('WaiterConfig', {})
        sleep_amount = config.get('Delay', self.config.delay)
        max_attempts = config.get('MaxAttempts', self.config.max_attempts)
        last_matched_acceptor = None
        num_attempts = 0

        while True:
            response = await self._operation_method(**kwargs)
            num_attempts += 1
            for acceptor in acceptors:
                if acceptor.matcher_func(response):
                    last_matched_acceptor = acceptor
                    current_state = acceptor.state
                    break
            else:
                # If none of the acceptors matched, we should
                # transition to the failure state if an error
                # response was received.
                if is_valid_waiter_error(response):
                    # Transition to a failure state, which we
                    # can just handle here by raising an exception.
                    raise WaiterError(
                        name=self.name,
                        reason='An error occurred (%s): %s'
                        % (
                            response['Error'].get('Code', 'Unknown'),
                            response['Error'].get('Message', 'Unknown'),
                        ),
                        last_response=response,
                    )
            if current_state == 'success':
                logger.debug(
                    "Waiting complete, waiter matched the " "success state."
                )
                return response
            if current_state == 'failure':
                reason = 'Waiter encountered a terminal failure state: %s' % (
                    acceptor.explanation
                )
                raise WaiterError(
                    name=self.name,
                    reason=reason,
                    last_response=response,
                )
            if num_attempts >= max_attempts:
                if last_matched_acceptor is None:
                    reason = 'Max attempts exceeded'
                else:
                    reason = (
                        'Max attempts exceeded. Previously accepted state: %s'
                        % (acceptor.explanation)
                    )
                raise WaiterError(
                    name=self.name,
                    reason=reason,
                    last_response=response,
                )
            await asyncio.sleep(sleep_amount)