Sqs

async_boto.clients.sqs

logger module-attribute

logger = getLogger(__name__)

T module-attribute

T = TypeVar('T', bound=BaseModel)

AsyncSQSClient

AsyncSQSClient(aws_session)

Bases: BaseClient

Source code in async_boto/clients/sqs.py

def __init__(self, aws_session: boto3.Session | AsyncAWSSession):
    super().__init__(aws_session=aws_session, service_name="sqs")
    self._url = f"https://sqs.{self._aws_session.region_name}.amazonaws.com"

add_permission async

add_permission(request)

Source code in async_boto/clients/sqs.py

async def add_permission(
    self, request: AddPermissionRequest
) -> AddPermissionResponse:
    return await self._make_request(
        "AmazonSQS.AddPermission",
        request,
        AddPermissionResponse,
    )

cancel_message_move_task async

cancel_message_move_task(request)

Source code in async_boto/clients/sqs.py

async def cancel_message_move_task(
    self, request: CancelMessageMoveTaskRequest
) -> CancelMessageMoveTaskResponse:
    return await self._make_request(
        "AmazonSQS.CancelMessageMoveTask",
        request,
        CancelMessageMoveTaskResponse,
    )

change_message_visibility async

change_message_visibility(request)

Source code in async_boto/clients/sqs.py

async def change_message_visibility(
    self, request: ChangeMessageVisibilityRequest
) -> ChangeMessageVisibilityResponse:
    return await self._make_request(
        "AmazonSQS.ChangeMessageVisibility",
        request,
        ChangeMessageVisibilityResponse,
    )

change_message_visibility_batch async

change_message_visibility_batch(request)

Source code in async_boto/clients/sqs.py

async def change_message_visibility_batch(
    self, request: ChangeMessageVisibilityBatchRequest
) -> ChangeMessageVisibilityBatchResponse:
    return await self._make_request(
        "AmazonSQS.ChangeMessageVisibilityBatch",
        request,
        ChangeMessageVisibilityBatchResponse,
    )

create_queue async

create_queue(request)

Source code in async_boto/clients/sqs.py

async def create_queue(self, request: CreateQueueRequest) -> CreateQueueResponse:
    return await self._make_request(
        "AmazonSQS.CreateQueue",
        request,
        CreateQueueResponse,
    )

delete_message async

delete_message(request)

Source code in async_boto/clients/sqs.py

async def delete_message(
    self, request: DeleteMessageRequest
) -> DeleteMessageResponse:
    return await self._make_request(
        "AmazonSQS.DeleteMessage",
        request,
        DeleteMessageResponse,
    )

delete_message_batch async

delete_message_batch(request)

Source code in async_boto/clients/sqs.py

async def delete_message_batch(
    self, request: DeleteMessageBatchRequest
) -> DeleteMessageBatchResponse:
    return await self._make_request(
        "AmazonSQS.DeleteMessageBatch",
        request,
        DeleteMessageBatchResponse,
    )

delete_queue async

delete_queue(request)

Source code in async_boto/clients/sqs.py

async def delete_queue(self, request: DeleteQueueRequest) -> DeleteQueueResponse:
    return await self._make_request(
        "AmazonSQS.DeleteQueue",
        request,
        DeleteQueueResponse,
    )

get_queue_attributes async

get_queue_attributes(request)

Source code in async_boto/clients/sqs.py

async def get_queue_attributes(
    self, request: GetQueueAttributesRequest
) -> GetQueueAttributesResponse:
    return await self._make_request(
        "AmazonSQS.GetQueueAttributes",
        request,
        GetQueueAttributesResponse,
    )

get_queue_url async

get_queue_url(request)

Source code in async_boto/clients/sqs.py

async def get_queue_url(self, request: GetQueueUrlRequest) -> GetQueueUrlResponse:
    return await self._make_request(
        "AmazonSQS.GetQueueUrl",
        request,
        GetQueueUrlResponse,
    )

list_dead_letter_source_queues async

list_dead_letter_source_queues(request)

Source code in async_boto/clients/sqs.py

async def list_dead_letter_source_queues(
    self, request: ListDeadLetterSourceQueuesRequest
) -> ListDeadLetterSourceQueuesResponse:
    return await self._make_request(
        "AmazonSQS.ListDeadLetterSourceQueues",
        request,
        ListDeadLetterSourceQueuesResponse,
    )

list_message_move_tasks async

list_message_move_tasks(request)

Source code in async_boto/clients/sqs.py

async def list_message_move_tasks(
    self, request: ListMessageMoveTasksRequest
) -> ListMessageMoveTasksResponse:
    return await self._make_request(
        "AmazonSQS.ListMessageMoveTasks",
        request,
        ListMessageMoveTasksResponse,
    )

list_queues async

list_queues(request)

Source code in async_boto/clients/sqs.py

async def list_queues(self, request: ListQueuesRequest) -> ListQueuesResponse:
    return await self._make_request(
        "AmazonSQS.ListQueues",
        request,
        ListQueuesResponse,
    )

list_queue_tags async

list_queue_tags(request)

Source code in async_boto/clients/sqs.py

async def list_queue_tags(
    self, request: ListQueueTagsRequest
) -> ListQueueTagsResponse:
    return await self._make_request(
        "AmazonSQS.ListQueueTags",
        request,
        ListQueueTagsResponse,
    )

purge_queue async

purge_queue(request)

Source code in async_boto/clients/sqs.py

async def purge_queue(self, request: PurgeQueueRequest) -> PurgeQueueResponse:
    return await self._make_request(
        "AmazonSQS.PurgeQueue",
        request,
        PurgeQueueResponse,
    )

receive_message async

receive_message(request)

Source code in async_boto/clients/sqs.py

async def receive_message(
    self, request: ReceiveMessageRequest
) -> ReceiveMessageResponse:
    return await self._make_request(
        "AmazonSQS.ReceiveMessage",
        request,
        ReceiveMessageResponse,
    )

remove_permission async

remove_permission(request)

Source code in async_boto/clients/sqs.py

async def remove_permission(
    self, request: RemovePermissionRequest
) -> RemovePermissionResponse:
    return await self._make_request(
        "AmazonSQS.RemovePermission",
        request,
        RemovePermissionResponse,
    )

send_message async

send_message(request)

Source code in async_boto/clients/sqs.py

async def send_message(self, request: SendMessageRequest) -> SendMessageResponse:
    return await self._make_request(
        "AmazonSQS.SendMessage",
        request,
        SendMessageResponse,
    )

send_message_batch async

send_message_batch(request)

Source code in async_boto/clients/sqs.py

async def send_message_batch(
    self, request: SendMessageBatchRequest
) -> SendMessageBatchResponse:
    return await self._make_request(
        "AmazonSQS.SendMessageBatch",
        request,
        SendMessageBatchResponse,
    )

set_queue_attributes async

set_queue_attributes(request)

Source code in async_boto/clients/sqs.py

async def set_queue_attributes(
    self, request: SetQueueAttributesRequest
) -> SetQueueAttributesResponse:
    return await self._make_request(
        "AmazonSQS.SetQueueAttributes",
        request,
        SetQueueAttributesResponse,
    )

start_message_move_task async

start_message_move_task(request)

Source code in async_boto/clients/sqs.py

async def start_message_move_task(
    self, request: StartMessageMoveTaskRequest
) -> StartMessageMoveTaskResponse:
    return await self._make_request(
        "AmazonSQS.StartMessageMoveTask",
        request,
        StartMessageMoveTaskResponse,
    )

tag_queue async

tag_queue(request)

Source code in async_boto/clients/sqs.py

async def tag_queue(self, request: TagQueueRequest) -> TagQueueResponse:
    return await self._make_request(
        "AmazonSQS.TagQueue",
        request,
        TagQueueResponse,
    )

untag_queue async

untag_queue(request)

Source code in async_boto/clients/sqs.py

async def untag_queue(self, request: UntagQueueRequest) -> UntagQueueResponse:
    return await self._make_request(
        "AmazonSQS.UntagQueue",
        request,
        UntagQueueResponse,
    )

paginate async

paginate(method_name, request)

Source code in async_boto/core/base_client.py

async def paginate(self, method_name, request: BaseModel):
    if method_name not in self._paginators:
        raise ValueError(
            f"Method {method_name} is not paginatable. "
            f"Available methods: {list(self._paginators.keys())}"
        )
    paginator = paginate(self, request=request, **self._paginators[method_name])
    async for page in paginator:
        yield page