HEX

File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/SQS.cpython-39.pyc
a

X>h?��@s�dZddlmZddlZddlZddlZddlZddlmZddlm	Z	ddl
mZddlm
Z
ddlmZmZmZdd	lmZdd
lmZmZddlmZddlmZdd
lmZddlmZddlm Z m!Z!ddl"m#Z#m$Z$ddl%m&Z&ddl'm(Z(ee)�Z*dd�ej+D�Z,de,d<dZ-dd�Z.Gdd�de/�Z0Gdd�de/�Z1Gdd �d e/�Z2Gd!d"�d"e/�Z3Gd#d$�d$e(j4�Z4Gd%d&�d&e(j5�Z5Gd'd(�d(e(j6�Z6dS))a?Amazon SQS transport module for Kombu.

This package implements an AMQP-like interface on top of Amazons SQS service,
with the goal of being optimized for high performance and reliability.

The default settings for this module are focused now on high performance in
task queue situations where tasks are small, idempotent and run very fast.

SQS Features supported by this transport
========================================
Long Polling
------------
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html

Long polling is enabled by setting the `wait_time_seconds` transport
option to a number > 1.  Amazon supports up to 20 seconds.  This is
enabled with 10 seconds by default.

Batch API Actions
-----------------
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-batch-api.html

The default behavior of the SQS Channel.drain_events() method is to
request up to the 'prefetch_count' messages on every request to SQS.
These messages are stored locally in a deque object and passed back
to the Transport until the deque is empty, before triggering a new
API call to Amazon.

This behavior dramatically speeds up the rate that you can pull tasks
from SQS when you have short-running tasks (or a large number of workers).

When a Celery worker has multiple queues to monitor, it will pull down
up to 'prefetch_count' messages from queueA and work on them all before
moving on to queueB.  If queueB is empty, it will wait up until
'polling_interval' expires before moving back and checking on queueA.

Message Attributes
-----------------
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html

SQS supports sending message attributes along with the message body.
To use this feature, you can pass a 'message_attributes' as keyword argument
to `basic_publish` method.

Other Features supported by this transport
==========================================
Predefined Queues
-----------------
The default behavior of this transport is to use a single AWS credential
pair in order to manage all SQS queues (e.g. listing queues, creating
queues, polling queues, deleting messages).

If it is preferable for your environment to use multiple AWS credentials, you
can use the 'predefined_queues' setting inside the 'transport_options' map.
This setting allows you to specify the SQS queue URL and AWS credentials for
each of your queues. For example, if you have two queues which both already
exist in AWS) you can tell this transport about them as follows:

.. code-block:: python

    transport_options = {
      'predefined_queues': {
        'queue-1': {
          'url': 'https://sqs.us-east-1.amazonaws.com/xxx/aaa',
          'access_key_id': 'a',
          'secret_access_key': 'b',
          'backoff_policy': {1: 10, 2: 20, 3: 40, 4: 80, 5: 320, 6: 640}, # optional
          'backoff_tasks': ['svc.tasks.tasks.task1'] # optional
        },
        'queue-2.fifo': {
          'url': 'https://sqs.us-east-1.amazonaws.com/xxx/bbb.fifo',
          'access_key_id': 'c',
          'secret_access_key': 'd',
          'backoff_policy': {1: 10, 2: 20, 3: 40, 4: 80, 5: 320, 6: 640}, # optional
          'backoff_tasks': ['svc.tasks.tasks.task2'] # optional
        },
      }
    'sts_role_arn': 'arn:aws:iam::<xxx>:role/STSTest', # optional
    'sts_token_timeout': 900 # optional
    }

Note that FIFO and standard queues must be named accordingly (the name of
a FIFO queue must end with the .fifo suffix).

backoff_policy & backoff_tasks are optional arguments. These arguments
automatically change the message visibility timeout, in order to have
different times between specific task retries. This would apply after
task failure.

AWS STS authentication is supported, by using sts_role_arn, and
sts_token_timeout. sts_role_arn is the assumed IAM role ARN we are trying
to access with. sts_token_timeout is the token timeout, defaults (and minimum)
to 900 seconds. After the mentioned period, a new token will be created.



If you authenticate using Okta_ (e.g. calling |gac|_), you can also specify
a 'session_token' to connect to a queue. Note that those tokens have a
limited lifetime and are therefore only suited for short-lived tests.

.. _Okta: https://www.okta.com/
.. _gac: https://github.com/Nike-Inc/gimme-aws-creds#readme
.. |gac| replace:: ``gimme-aws-creds``


Client config
-------------
In some cases you may need to override the botocore config. You can do it
as follows:

.. code-block:: python

    transport_option = {
      'client-config': {
          'connect_timeout': 5,
       },
    }

For a complete list of settings you can adjust using this option see
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html

Features
========
* Type: Virtual
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: Yes
* Supports Priority: No
* Supports TTL: No
�)�annotationsN)�datetime)�Empty)�Config)�ClientError)�ensure_promise�promise�	transform)�get_event_loop)�boto3�
exceptions)�AsyncSQSConnection)�AsyncMessage)�
get_logger)�
scheduling)�bytes_to_str�safe_str)�dumps�loads)�cached_property�)�virtualcCsi|]}|dvrt|�d�qS)z-_.�_)�ord)�.0�c�r�=/usr/local/lib/python3.9/site-packages/kombu/transport/SQS.py�
<dictcomp>�sr�-�.�
cCs&z
t|�WSty |YS0dS)z5Try to convert x' to int, or return x' if that fails.N)�int�
ValueError)�xrrr�	maybe_int�s
r%c@seZdZdZdS)�UndefinedQueueExceptionzAPredefined queues are being used and an undefined queue was used.N��__name__�
__module__�__qualname__�__doc__rrrrr&�sr&c@seZdZdZdS)�InvalidQueueExceptionz@Predefined queues are being used and configuration is not valid.Nr'rrrrr,�sr,c@seZdZdZdS)�AccessDeniedQueueExceptionz�Raised when access to the AWS queue is denied.

    This may occur if the permissions are not correctly set or the
    credentials are invalid.
    Nr'rrrrr-�sr-c@seZdZdZdS)�DoesNotExistQueueExceptionz"The specified queue doesn't exist.Nr'rrrrr.�sr.cs:eZdZdZd�fdd�	Zdd�Zdd�Zd	d
�Z�ZS)�QoSz5Quality of Service guarantees implementation for SQS.FcsFt�j||d�|�|�\}}}}|rB|rB|rB|rB|�||||�dS)N)�requeue)�super�reject�1_extract_backoff_policy_configuration_and_message�apply_backoff_policy)�self�delivery_tagr0�routing_key�message�
backoff_tasks�backoff_policy��	__class__rrr2�s��
�z
QoS.rejectcCsjz|j|}|jd}Wnty,YdS0|r6|s:dS|jj�|i�}|�d�}|�d�}||||fS)Nr7)NNNNr9r:)�
_delivered�
delivery_info�KeyError�channel�predefined_queues�get)r5r6r8r7Zqueue_configr9r:rrrr3�s


z5QoS._extract_backoff_policy_configuration_and_messagec
Cs`|jj|}|�|�\}}|r"|s&dS|�|�}||vr\|dur\|j�|�}	|	j|||d�dS)N��QueueUrl�
ReceiptHandle�VisibilityTimeout)r@�_queue_cache�'extract_task_name_and_number_of_retriesrB�sqs�change_message_visibility)
r5r7r6r:r9�	queue_url�	task_name�number_of_retriesZpolicy_valuerrrrr4�s�
�zQoS.apply_backoff_policycCs:|j|}|j}|d}t|jdddd�}||fS)N�taskr>�sqs_message�
AttributesZApproximateReceiveCount)r=�headersr"�
properties)r5r6r8Zmessage_headersrLrMrrrrH�s
���z+QoS.extract_task_name_and_number_of_retries)F)	r(r)r*r+r2r3r4rH�
__classcell__rrr;rr/�s
	
r/csTeZdZdZdZdZdZdZdZiZ	dZ
iZiZe
�ZeZ�fdd�Zd	d
�Zdd�Z�fd
d�Z�fdd�Zdpdd�Zdd�Zefdd�Zdd�Zdd�Zdd�Zdd�Zdd �Zd!d"�Zed#d$��Z d%d&�Z!d'd(�Z"e#dfd)d*�Z$d+d,�Z%dqd-d.�Z&d/d0�Z'e#fd1d2�Z(e#dfd3d4�Z)drd6d7�Z*d8d9�Z+dsd:d;�Z,dt�fd=d>�	Z-du�fd@dA�	Z.dBdC�Z/dDdE�Z0�fdFdG�Z1dvdHdI�Z2dwdJdK�Z3dLdM�Z4dNdO�Z5dPdQ�Z6dxdRdS�Z7e8dTdU��Z9e8dVdW��Z:e;dXdY��Z<e;dZd[��Z=e;d\d]��Z>e;d^d_��Z?e;d`da��Z@e;dbdc��ZAe;ddde��ZBe;dfdg��ZCe;dhdi��ZDe;djdk��ZEe;dldm��ZFe;dndo��ZG�ZHS)y�ChannelzSQS Channel.z	us-east-1ir!zkombu%(vhost)sNcsLtdurtd��t�j|i|��|��|�|j�|�d�pDt�|_	dS)Nzboto3 is not installed�hub)
r�ImportErrorr1�__init__�_validate_predifined_queues�_update_queue_cache�queue_name_prefixrBr
rU)r5�args�kwargsr;rrrW	szChannel.__init__cCsh|j��D]X\}}|d�d�}|�d�}|rF|sFtd�|d���q
|s
|r
td�||d���q
dS)z�Check that standard and FIFO queues are named properly.

        AWS requires FIFO queues to have a name
        that ends with the .fifo suffix.
        �url�.fifoz6Queue with url '{}' must have a name ending with .fifoz.Queue with name '{}' is not a FIFO queue: '{}'N)rA�items�endswithr,�format)r5�
queue_name�qZfifo_urlZ	fifo_namerrrrXs
����z#Channel._validate_predifined_queuescCsj|jr,|j��D]\}}|d|j|<qdS|��j|d�}|�dg�D]}|�d�d}||j|<qHdS)Nr])ZQueueNamePrefixZ	QueueUrls�/���)rAr_rGrIZlist_queuesrB�split)r5rZrbrc�respr]rrrrY+szChannel._update_queue_cachecs<|r|j�|�|jr |�|�t�j||g|�Ri|��S�N)�
_noack_queues�addrU�_loop1r1�
basic_consume)r5�queueZno_ackr[r\r;rrrl6s
���zChannel.basic_consumecs,||jvr |j|}|j�|�t��|�Srh)�
_consumersZ
_tag_to_queueri�discardr1�basic_cancel)r5Zconsumer_tagrmr;rrrp?s

zChannel.basic_cancelcKs,|jr|j��st��|j|j||d�dS)z�Return a single payload message from one of our queues.

        Raises
        ------
            Queue.Empty: if no messages available.
        )�timeoutN)rn�qos�can_consumerZ_poll�cycle)r5rq�callbackr\rrr�drain_eventsEszChannel.drain_eventscCst�|j|jt�|_dS)aAReset the consume cycle.

        Returns
        -------
            FairCycle: object that points to our _get_bulk() method
                rather than the standard _get() method.  This allows for
                multiple messages to be returned at once from SQS (
                based on the prefetch limit).
        N)rZ	FairCycle�	_get_bulk�_active_queuesrZ_cycle�r5rrr�_reset_cycleSs

�zChannel._reset_cyclecCsL|�d�r6|dtd��}tt|���|�}|dStt|���|�SdS)z3Format AMQP queue name into a legal SQS queue name.r^N)r`�len�strr�	translate)r5�name�table�partialrrr�entity_nameas

zChannel.entity_namecCs|�|j|�Srh)r�rZ)r5rbrrr�canonical_queue_namejszChannel.canonical_queue_namecCsf|�|�}||jvr|�|�z|j|WSty`|jrLtd�|���td|�d���Yn0dS)z9Try to retrieve the SQS queue URL for a given queue name.�<Queue with name '{}' must be defined in 'predefined_queues'.�Queue with name 'z' doesn't exist in SQSN)r�rGrYr?rAr&rar.)r5rm�	sqs_qnamerrr�_resolve_queue_urlms


��
�zChannel._resolve_queue_urlcKspz|�|�WStyj|�|�}dt|j�i}|�d�rDd|d<|�||�}|d|j|<|dYS0dS)z�Ensure a queue with given name exists in SQS.

        Arguments:
        ---------
            queue (str): the AMQP queue name
        Returns
            str: the SQS queue URL
        rFr^�trueZ	FifoQueuerDN)r�r.r�r|�visibility_timeoutr`�
_create_queuerG)r5rmr\r��
attributesrgrrr�
_new_queue�s	

zChannel._new_queuecCs6|jr
dS|�|j�d�pi�|j|d�j||d�S)z=Create an SQS queue with a given name and nominal attributes.Nzsqs-creation-attributes�rm)Z	QueueNamerP)rA�update�transport_optionsrBrIZcreate_queue)r5rbr�rrrr��s��zChannel._create_queuecOs6|jr
dS|�|�}|��j|d�|j�|d�dS)zDelete queue by name.N�rD)rAr�rIZdelete_queuerG�pop)r5rmr[r\�q_urlrrr�_delete�s
�zChannel._deletecKs(|�|�}d|i}d|vr�d|dvr8|d�d�|d<|�d�r�d|dvr`|dd|d<nd|d<d|dvr�|dd|d<q�tt���|d<nd	|dvr�|dd	|d	<|jr�t��t	|��}nt	|�}||d
<|j
|�|�d�}|�d��r|j
||dd
dd�n|jfi|��dS)zPut message onto queue.rDrRZmessage_attributesZMessageAttributesr^ZMessageGroupId�defaultZMessageDeduplicationIdZDelaySecondsZMessageBodyr�Zredeliveredr6rrCN)r�r�r`r|�uuid�uuid4�sqs_base64_encodingr�encoderrIr�rBrJ�send_message)r5rmr8r\r��bodyrrrr�_put�s<
�

�
�
�
�zChannel._putcCs:z"t�|�}t�|�|kr |WSWnty4Yn0|Srh)�base64�	b64decode�	b64encode�	Exception)Zbyte_string�datarrr�_optional_b64_decode�s

zChannel._optional_b64_decodecCs�|�|d���}tt|��}||jvrL|�|�}|j|d��||d�nhz|d}|dd}Wn2ty�i}d|i}|�	t|�|d��Yn0|�	||d��|d|d<|S)	N�Bodyr�rErRr>)r�rR�rO�	sqs_queuer6)
r�r�rrrir��asynsqs�delete_messager?r�)r5r8rbr�r��payloadrRr>rrr�_message_to_python�s.

���zChannel._message_to_pythoncs ��������fdd�|D�S)a�Convert a list of SQS Message objects into Payloads.

        This method handles converting SQS Message objects into
        Payloads, and appropriately updating the queue depending on
        the 'ack' settings for that queue.

        Arguments:
        ---------
            messages (SQSMessage): A list of SQS Message objects.
            queue (str): Name representing the queue they came from.

        Returns
        -------
            List: A list of Payload objects
        csg|]}��|����qSr)r�)r�m�r�rmr5rr�
<listcomp>�z/Channel._messages_to_python.<locals>.<listcomp>)r�)r5�messagesrmrr�r�_messages_to_pythons
zChannel._messages_to_pythonc	Cs�|��}|r�|�|�}|j|d�j|||jd�}|�d�r�|dD]}t|dd���|d<qB|�|d|�D]}|j	�
||�qndSt��dS)a;Try to retrieve multiple messages off ``queue``.

        Where :meth:`_get` returns a single Payload object, this method
        returns a list of Payload objects.  The number of objects returned
        is determined by the total number of messages available in the queue
        and the number of messages the QoS object allows (based on the
        prefetch_count).

        Note:
        ----
            Ignores QoS limits so caller is responsible for checking
            that we are allowed to consume at least one message from the
            queue.  get_bulk will then ask QoS for an estimate of
            the number of extra messages that we can consume.

        Arguments:
        ---------
            queue (str): The queue name to pull from.

        Returns
        -------
            List[Message]
        r��rDZMaxNumberOfMessagesZWaitTimeSeconds�Messagesr��r�N)�_get_message_estimater�rI�receive_message�wait_time_secondsrBr�decoder��
connectionZ_deliverr)	r5rm�max_if_unlimitedru�	max_countr�rgr��msgrrrrws
�
zChannel._get_bulkcCsv|�|�}|j|d�j|d|jd�}|�d�rlt|dddd���}||ddd<|�|d|�dSt��dS)	z/Try to retrieve a single message off ``queue``.r�rr�r�rr�r�N)	r�rIr�r�rBrr�r�r)r5rmr�rgr�rrr�_getFs
�
zChannel._getcCs|j�|j|�dSrh)rU�	call_soon�_schedule_queue)r5rm�_rrrrkRszChannel._loop1cCs<||jvr8|j��r.|j|t|j|f�d�n
|�|�dS�N)ru)rxrrrs�_get_bulk_asyncrrk)r5rmrrrr�Us

�zChannel._schedule_queuecCs&|j��}t|dur|nt|d�|�S)Nr)rrZcan_consume_max_estimate�min�max)r5r��maxcountrrrr�^s

�zChannel._get_message_estimatecCs0|��}|r|j|||d�St|�}|g�|Sr�)r��
_get_asyncr)r5rmr�rur�rrrr�eszChannel._get_bulk_asyncrc
Cs<|�|�}|�|�}|j||||j|d�t|j|||�d�S)Nr�)rbrK�countr�ru)r�r��
_get_from_sqsr�r	�_on_messages_ready)r5rmr�rur��qnamerrrr�os



��zChannel._get_asynccCsDd|vr@|dr@|jj}|dD]}|�|||�}|||�q dS)Nr�)r��
_callbacksr�)r5rmr�r��	callbacksr�Z
msg_parsedrrrr�zs
zChannel._on_messages_readycCs|j||||j|d�S)zwRetrieve and handle messages from SQS.

        Uses long polling and returns :class:`~vine.promises.promise`.
        )Znumber_messagesr�ru)r�r�)r5rbrKr�r�rurrrr��s
�zChannel._get_from_sqsr�cs$|D]}|j�|d�qt��|�Srh)r>r�r1�_restore)r5r8Zunwanted_delivery_infoZunwanted_keyr;rrr��szChannel._restoreFc
s�z|j�|�j}|d}Wnty8t��|�Yn�0d}d|vrT|�|d�}z"|j|d�j|d|dd�WnTt	y�}z<|j
ddd	kr�t|j
dd
��t��|�WYd}~nd}~00t��|�dS)NrOr7r�r�rE)rDrE�ErrorZCodeZAccessDenied�Message)
rrrBr>r?r1�	basic_ackr�rIr�r�responser-Zbasic_reject)r5r6�multipler8rOrm�	exceptionr;rrr��s(�
�"zChannel.basic_ackcCs<|�|�}|j|�|�d�}|j|dgd�}t|dd�S)z)Return the number of messages in a queue.r�ZApproximateNumberOfMessages)rDZAttributeNamesrP)r�rIr�Zget_queue_attributesr")r5rmr�rrgrrr�_size�s
�z
Channel._sizecCsN|�|�}d}td�D]}|t|�|��7}|sq6q|j|d�j|d�|S)z'Delete all current messages in a queue.rr!r�r�)r��ranger"r�rIZpurge_queue)r5rmr��size�irrr�_purge�s
zChannel._purgecst���dSrh)r1�closeryr;rrr��sz
Channel.closec
Csvtjj||||d�}|jdur$|jnd}d|i}|jdurD|j|d<|j�d�pRi}tfi|��}	|jdd|	i|��S)	N)�region_nameZaws_access_key_idZaws_secret_access_keyZaws_session_tokenTZuse_ssl�endpoint_urlz
client-configrI�config)rI)	r�session�Session�	is_securer�r�rBr�client)
r5�region�
access_key_id�secret_access_key�
session_tokenr�r�Z
client_kwargsZ
client_configr�rrr�new_sqs_client�s��

zChannel.new_sqs_clientcCs�|dur�|jr�||jvr(td|�d���|j|}|j�d�rJ|�||�S|j�d�s�||jvrj|j|S|j|�d|j�|�d|jj	�|�d|jj
�d�}|j|<|S|jdur�|jS|j|j|jj	|jj
d�}|_|S)Nr�z)' must be defined in 'predefined_queues'.�sts_role_arnr�r�r�)r�r�r�)rAr&r�rB�_handle_sts_session�_predefined_queue_clientsr�r��conninfoZuserid�password�_sqs�r5rmrcrrrrrI�s<

�


����
�zChannel.sqscCsj|�d|j�}t|d�s$|�||�S|jjdd�t��krF|�||�S||jvr\|�||�S|j|SdS)Nr��sts_expiration)�tzinfo)	rBr��hasattr�-_new_predefined_queue_client_with_sts_sessionr��replacer�utcnowr�)r5rmrcr�rrrr�s

zChannel._handle_sts_sessioncCsT|�|j�d�|j�dd��}|d|_|j||d|d|dd�}|j|<|S)	Nr�Zsts_token_timeouti�Z
ExpirationZAccessKeyIdZSecretAccessKeyZSessionToken)r�r�r�r�)�generate_sts_session_tokenr�rBr�r�r�)r5rmr�Z	sts_credsrrrrr�s
�
�z5Channel._new_predefined_queue_client_with_sts_sessioncCs"t�d�}|j|d|d�}|dS)N�stsZCelery)ZRoleArnZRoleSessionNameZDurationSeconds�Credentials)rr�Zassume_role)r5Zrole_arnZtoken_expiry_secondsZ
sts_clientZ
sts_policyrrrr�s
�z"Channel.generate_sts_session_tokencCs�|dur~|jr~||jvr,t|d�s,|j|S||jvrDtd�|���|j|}t|j|d�|�d|j�|j	d�}|j|<|S|j
dur�|j
St|j|d�|j|j	d�}|_
|S)Nr�r�r�r�)Zsqs_connectionr��fetch_message_attributes)rA�_predefined_queue_async_clientsr�r&rar
rIrBr�r��_asynsqsr�rrrr�"s6
�

��

��

�zChannel.asynsqscCs|jjSrh)r�r�ryrrrr�?szChannel.conninfocCs
|jjjSrh)r�r�r�ryrrrr�CszChannel.transport_optionscCs|j�d�p|jS)Nr�)r�rB�default_visibility_timeoutryrrrr�Gs�zChannel.visibility_timeoutcCs|j�di�S)z/Map of queue_name to predefined queue settings.rA�r�rBryrrrrALszChannel.predefined_queuescCs|j�dd�S)NrZ�r�ryrrrrZQszChannel.queue_name_prefixcCsdS)NFrryrrr�supports_fanoutUszChannel.supports_fanoutcCs|j�d�pt��jp|jS)Nr�)r�rBrr�r��default_regionryrrrr�Ys
��zChannel.regioncCs|j�d�S)N�
regioninfor�ryrrrr�_szChannel.regioninfocCs|j�d�S)Nr�r�ryrrrr�cszChannel.is_securecCs|j�d�S�N�portr�ryrrrr�gszChannel.portcCsP|jjdurL|jrdnd}|jjdur6d|jj��}nd}d�||jj|�SdS)N�https�http�:r�z	{}://{}{})r��hostnamer�r�ra)r5�schemer�rrrr�ks�zChannel.endpoint_urlcCs|j�d|j�S)Nr�)r�rB�default_wait_time_secondsryrrrr�ys�zChannel.wait_time_secondscCs|j�dd�S)Nr�Tr�ryrrrr�~szChannel.sqs_base64_encodingcCs|j�d�S)Nr�r�ryrrrr��sz Channel.fetch_message_attributes)NN)N)rN)rN)r�)F)N)N)N)Ir(r)r*r+r�r�rZ
domain_formatr�r�r�r�rG�setrir/rWrXrYrlrprvrz�CHARS_REPLACE_TABLEr�r�r�r�r�r�r��staticmethodr�r�r��SQS_MAX_MESSAGESrwr�rkr�r�r�r�r�r�r�r�r�r�r�r�rIr�r�r�r��propertyr�r�rr�rArZr�r�r�r�r�r�r�r�r�rSrrr;rrT�s�	
	)
�
,
	�


�

�	

�

"
	













rTc@speZdZdZeZdZdZdZej	j
eje
jfZ
ej	jejfZdZdZej	jjdedg�d�Zed	d
��ZdS)�	TransportajSQS Transport.

    Additional queue attributes can be supplied to SQS during queue
    creation by passing an ``sqs-creation-attributes`` key in
    transport_options. ``sqs-creation-attributes`` must be a dict whose
    key-value pairs correspond with Attributes in the
    `CreateQueue SQS API`_.

    For example, to have SQS queues created with server-side encryption
    enabled using the default Amazon Managed Customer Master Key, you
    can set ``KmsMasterKeyId`` Attribute. When the queue is initially
    created by Kombu, encryption will be enabled.

    .. code-block:: python

        from kombu.transport.SQS import Transport

        transport = Transport(
            ...,
            transport_options={
                'sqs-creation-attributes': {
                    'KmsMasterKeyId': 'alias/aws/sqs',
                },
            }
        )

    .. _CreateQueue SQS API: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_CreateQueue.html#API_CreateQueue_RequestParameters

    The ``ApproximateReceiveCount`` message attribute is fetched by this
    transport by default. Requested message attributes can be changed by
    setting ``fetch_message_attributes`` in the transport options.

    .. code-block:: python

        from kombu.transport.SQS import Transport

        transport = Transport(
            ...,
            transport_options={
                'fetch_message_attributes': ["All"],
            }
        )

    .. _Message Attributes: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html#SQS-ReceiveMessage-request-AttributeNames

    rrNrIT�direct)ZasynchronousZ
exchange_typecCs
d|jiSr�)�default_portryrrr�default_connection_params�sz#Transport.default_connection_params)r(r)r*r+rTZpolling_intervalr�r
rrZconnection_errorsrZ
BotoCoreError�socket�errorZchannel_errorsZdriver_typeZdriver_nameZ
implements�extend�	frozensetrrrrrrr�s&/
����r)7r+�
__future__rr�r�stringr�rrmrZbotocore.clientrZbotocore.exceptionsrZvinerrr	Zkombu.asynchronousr
Zkombu.asynchronous.aws.extrrZ%kombu.asynchronous.aws.sqs.connectionr
Z"kombu.asynchronous.aws.sqs.messagerZ	kombu.logrZkombu.utilsrZkombu.utils.encodingrrZkombu.utils.jsonrrZkombu.utils.objectsrr�rr(�logger�punctuationrrr%r�r&r,r-r.r/rTrrrrr�<module>sP�3