o wvXjd@sTdZddlmZmZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z ddlmZddlmZmZddlmZdd lmZdd lmZmZmZmZdd lmZdd lmZdd l m!Z!m"Z"ddl#m$Z$m%Z%ddl&m'Z'ddl(m)Z)ee*Z+ddej,DZ-de-d<dZ.ddZ/Gddde0Z1Gddde)j2Z2Gddde)j3Z3dS)aV Amazon SQS Transport. 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. 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 a single AWS credential, 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: transport_options = { 'predefined_queues': { 'queue-1': { 'url': 'https://sqs.us-east-1.amazonaws.com/xxx/aaa', 'access_key_id': 'a', 'secret_access_key': 'b', }, 'queue-2': { 'url': 'https://sqs.us-east-1.amazonaws.com/xxx/bbb', 'access_key_id': 'c', 'secret_access_key': 'd', }, } } )absolute_importunicode_literalsN) ClientError) transformensure_promisepromise)get_event_loop)boto3 exceptions)AsyncSQSConnection) AsyncMessage)Emptyrangestring_ttext_t) get_logger) scheduling) bytes_to_strsafe_str)loadsdumps)cached_property)virtualcCsi|] }|dvrt|dqS)z-_._)ord).0crP/var/www/html/myproject/venv/lib/python3.10/site-packages/kombu/transport/SQS.py Zsr -. cCs"zt|WSty|YSw)z5Try to convert x' to int, or return x' if that fails.)int ValueError)xrrr maybe_intcs   r'c@seZdZdZdS)UndefinedQueueExceptionzAPredefined queues are being used and an undefined queue was used.N)__name__ __module__ __qualname____doc__rrrrr(ksr(cs eZdZdZdZdZdZdZdZiZ dZ iZ iZ e ZfddZd d Zfd d Zfd dZd`ddZddZefddZddZddZddZfddZddZdd Zd!d"Zedfd#d$Zd%d&Z dad'd(Z!d)d*Z"efd+d,Z#edfd-d.Z$dbd0d1Z%d2d3Z& dcd4d5Z' 6ddfd7d8 Z(defd:d; Z)dd?Z+fd@dAZ,dBdCZ-dadDdEZ.dadFdGZ/e0dHdIZ1e0dJdKZ2e3dLdMZ4e3dNdOZ5e3dPdQZ6e3dRdSZ7e3dTdUZ8e3dVdWZ9e3dXdYZ:e3dZd[Z;e3d\d]ZS)fChannelz SQS Channel.z us-east-1ir#zkombu%(vhost)sNcsHtdurtdtt|j|i|||j|dp t|_ dS)Nzboto3 is not installedhub) r ImportErrorsuperr-__init___update_queue_cachequeue_name_prefixgetrr.)selfargskwargs __class__rrr1}s  zChannel.__init__cCsj|jr|jD] \}}|d|j|<qdS|j|d}|dgD]}|dd}||j|<q$dS)Nurl)QueueNamePrefix QueueUrls/)predefined_queuesitems _queue_cachesqs list_queuesr4split)r5r3 queue_nameqrespr:rrrr2s zChannel._update_queue_cachecs@|r|j||jr||tt|j||g|Ri|SN) _noack_queuesaddr._loop1r0r- basic_consume)r5queueno_ackr6r7r8rrrLs   zChannel.basic_consumecs0||jvr|j|}|j|tt||SrH) _consumers _tag_to_queuerIdiscardr0r- basic_cancel)r5 consumer_tagrMr8rrrRs   zChannel.basic_cancelcKs,|jr|js t|j|j||ddS)zReturn a single payload message from one of our queues. Raises: Queue.Empty: if no messages available. )timeoutN)rOqos can_consumer _pollcycle)r5rTcallbackr7rrr drain_eventsszChannel.drain_eventscCst|j|jt|_dS)a2Reset 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)r FairCycle _get_bulk_active_queuesr _cycler5rrr _reset_cycles  zChannel._reset_cyclecCsH|dr|dtd }tt||}|dStt||S)z3Format AMQP queue name into a legal SQS queue name..fifoN)endswithlenrr translate)r5nametablepartialrrr entity_names zChannel.entity_namecCs||j|SrH)rhr3)r5rErrrcanonical_queue_nameszChannel.canonical_queue_namecKst|ts|S||}||jvr||z|j|WStyP|jr-td|dt |j i}| dr=d|d<| ||}|d|j|<|dYSw)z-Ensure a queue with given name exists in SQS.Isz/Channel._messages_to_python..)rv)r5messagesrMrrr_messages_to_python:s zChannel._messages_to_pythonc Cs|}|rC||}|j|dj|||jd}|drC|dD] }t|dd|d<q!||d|D] }|j ||q7dSt )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] rwrnMaxNumberOfMessagesWaitTimeSecondsMessagesrrN) _get_message_estimatervrBreceive_messagewait_time_secondsr4r decoder connection_deliverr ) r5rMmax_if_unlimitedrY max_countrrGrmsgrrrr\Ks    zChannel._get_bulkcCsr||}|j|dj|d|jd}|dr6t|dddd}||ddd<||d|dSt)z/Try to retrieve a single message off ``queue``.rwrrrrrr) rvrBrrr4r rrr )r5rMrrGrrrr_getts   z Channel._getcCs|j|j|dSrH)r. call_soon_schedule_queue)r5rM_rrrrKszChannel._loop1cCsB||jvr|jr|j|t|j|fddS||dSdSN)rY)r]rUrV_get_bulk_asyncrrK)r5rMrrrrs   zChannel._schedule_queuecCs*|j}t|dur||St|d|S)Nr)rUcan_consume_max_estimateminmax)r5rmaxcountrrrrs  zChannel._get_message_estimatecCs0|}|r|j|||dSt|}|g|Sr)r _get_asyncr)r5rMrrYrrrrrs zChannel._get_bulk_asyncrc Cs:||}||}|j|||j|dt|j|||dS)Nrw)countrrY)rvri _get_from_sqsrr_on_messages_ready)r5rMrrYrFqnamerrrrs  zChannel._get_asynccCsLd|vr |dr"|jj}|dD]}||||}|||qdSdSdS)Nr)r _callbacksr)r5rMrr callbacksr msg_parsedrrrrs zChannel._on_messages_readycCs\|dur|n|j}|jr||jvrtd||j|}n||}|j||||j|dS)zwRetrieve and handle messages from SQS. Uses long polling and returns :class:`~vine.promises.promise`. Nrj)number_messagesrrY)rr?rAr(rq get_queue_urlrr)r5rMrrrY queue_urlrrrrs   zChannel._get_from_sqsrcs(|D] }|j|dqtt||SrH)rr~r0r-_restore)r5runwanted_delivery_info unwanted_keyr8rrrszChannel._restoreFcsz |j|j}|d}Wntytt||YdSwd}d|vr-||d}z|j|dj |d|ddWnt yPtt| |YdSwtt||dS)Nr routing_keyrwrr)rnr) rUr4rrpr0r- basic_ackrirBrr basic_reject)r5rmultiplerrrMr8rrrs$    zChannel.basic_ackcCs<||}|j||d}|j|dgd}t|ddS)z)Return the number of messages in a queue.rwApproximateNumberOfMessages)rnAttributeNamesry)rvrBriget_queue_attributesr$)r5rMr:rrGrrr_sizes z Channel._sizecCsN||}d}tdD]}|t||7}|snq |j|dj|d|S)z'Delete all current messages in a queue.rr#rw)rn)rvrr$rrB purge_queue)r5rMrFsizeirrr_purges  zChannel._purgecstt|dSrH)r0r-closer_r8rrrsz Channel.closecCsRtjj|||d}|jdur|jnd}d|i}|jdur!|j|d<|jdi|S)N) region_nameaws_access_key_idaws_secret_access_keyTuse_ssl endpoint_urlrB)rB)r sessionSession is_securerclient)r5region access_key_idsecret_access_keyrr client_kwargsrrrnew_sqs_clients  zChannel.new_sqs_clientcCs|durB|jrB||jvr|j|S||jvrtd||j|}|j|d|j|d|jj|d|jj d}|j|<|S|j durJ|j S|j|j|jj|jj d}|_ |S)Nrjrrr)rrr) r?_predefined_queue_clientsr(rqrr4rconninfouseridpassword_sqsr5rMrFrrrrrB s.       z Channel.sqscCs|dur8|jr8||jvr|j|S||jvrtd||j|}t|j|d|d|jd}|j|<|S|jdur@|jSt|j|d|jd}|_|S)Nrjrwr)sqs_connectionr) r?_predefined_queue_async_clientsr(rqr rBr4r_asynsqsrrrrr&s*         zChannel.asynsqscCs|jjSrH)rrr_rrrr?szChannel.conninfocCs |jjjSrH)rrr{r_rrrr{C zChannel.transport_optionscCs|jdp|jS)Nrs)r{r4default_visibility_timeoutr_rrrrsGs zChannel.visibility_timeoutcC|jddS)z/Map of queue_name to predefined queue settings.r?Nr{r4r_rrrr?LszChannel.predefined_queuescCr)Nr3rr_rrrr3QszChannel.queue_name_prefixcCsdS)NFrr_rrrsupports_fanoutUszChannel.supports_fanoutcCs|jdp tjp |jS)Nr)r{r4r rrdefault_regionr_rrrrYs zChannel.regioncC |jdS)N regioninforr_rrrr_ zChannel.regioninfocCr)Nrrr_rrrrcrzChannel.is_securecCrNportrr_rrrrgrz Channel.portcCsP|jjdur&|jr dnd}|jjdurd|jj}nd}d||jj|SdS)Nhttpshttpz:{}rz {}://{}{})rhostnamerrrq)r5schemerrrrrks  zChannel.endpoint_urlcCs|jd|jS)Nr)r{r4default_wait_time_secondsr_rrrryszChannel.wait_time_seconds)NNrH)rN)rNN)r)F)?r)r*r+r,rrr domain_formatrrrrrAsetrIr1r2rLrRrZr`CHARS_REPLACE_TABLErhrirvrtr}rrrSQS_MAX_MESSAGESr\rrKrrrrrrrrrrrrrBrpropertyrr{rrsr?r3rrrrrrr __classcell__rrr8rr-os      )                   r-c@speZdZdZeZdZdZdZej j e j e jfZ ej je j fZdZdZej jjdedgdZed d ZdS) TransportaSQS 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 rrNrBTdirect) asynchronous exchange_typecCs d|jiSr) default_portr_rrrdefault_connection_paramsrz#Transport.default_connection_params)r)r*r+r,r-polling_intervalrrrrconnection_errorsr BotoCoreErrorsocketerrorchannel_errors driver_type driver_name implementsextend frozensetrrrrrrrs( r)4r, __future__rrrr stringrbotocore.exceptionsrvinerrrkombu.asynchronousrkombu.asynchronous.aws.extr r %kombu.asynchronous.aws.sqs.connectionr "kombu.asynchronous.aws.sqs.messager kombu.fiver rrr kombu.logr kombu.utilsrkombu.utils.encodingrrkombu.utils.jsonrrkombu.utils.objectsrrrr)logger punctuationrrr' Exceptionr(r-rrrrrsB>