^fVdZddlmZddlZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZmZmZdd lmZdd lmZmZdd lmZdd lmZdd lmZddlmZddlm Z m!Z!ddl"m#Z#m$Z$ddl%m&Z&ddl'm(Z(ee)Z*dej+DZ,de,d<dZ-dZ.Gdde/Z0Gdde/Z1Gdde/Z2Gdd e(j3Z3Gd!d"e(j4Z4Gd#d$e(j5Z5dS)%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:::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_promisepromise transform)get_event_loop)boto3 exceptions)AsyncSQSConnection) AsyncMessage) get_logger) scheduling) bytes_to_strsafe_str)dumpsloads)cached_property)virtualc6i|]}|dvt|dS)z-_._)ord).0cs E/var/www/html/env/lib/python3.11/site-packages/kombu/transport/SQS.py rs,CFFD-. cF t|S#t$r|cYSwxYw)z5Try to convert x' to int, or return x' if that fails.)int ValueError)xs r maybe_intr(s61vv s   ceZdZdZdS)UndefinedQueueExceptionzAPredefined queues are being used and an undefined queue was used.N__name__ __module__ __qualname____doc__r rr*r*sKKKKr r*ceZdZdZdS)InvalidQueueExceptionz@Predefined queues are being used and configuration is not valid.Nr+r0r rr2r2sJJJJr r2ceZdZdZdS)AccessDeniedQueueExceptionzRaised when access to the AWS queue is denied. This may occur if the permissions are not correctly set or the credentials are invalid. Nr+r0r rr4r4sr r4c6eZdZdZdfd ZdZdZdZxZS)QoSz5Quality of Service guarantees implementation for SQS.Fct||||\}}}}|r |r |r |r |||||dSdSdSdSdS)N)requeue)superreject1_extract_backoff_policy_configuration_and_messageapply_backoff_policy)self delivery_tagr8 routing_keymessage backoff_tasksbackoff_policy __class__s rr:z QoS.rejects |W555  B B   < Wm^  J7 J} J J  % %\>= J J J J J J J J J J J J Jr c |j|}|jd}n#t$rYdSwxYw|r|sdS|jj|i}|d}|d}||||fS)Nr?)NNNNrArB) _delivered delivery_infoKeyErrorchannelpredefined_queuesget)r=r>r@r? queue_configrArBs rr;z5QoS._extract_backoff_policy_configuration_and_messages *ol3G!/ >KK * * *))) * *' *))|599+rJJ $((99 %))*:;;G]NBBs  ++c|jj|}||\}}|r|sdS||}||vr6|6|j|} | |||dSdSdS)NQueueUrl ReceiptHandleVisibilityTimeout)rH _queue_cache'extract_task_name_and_number_of_retriesrJsqschange_message_visibility) r=r?r>rBrA queue_url task_namenumber_of_retries policy_valuers rr<zQoS.apply_backoff_policysL-k:  8 8 F F % $  1 4%))*;<<  % %,*B   --A ' '"*". (      & %*B*Br c|j|}|j}|d}t|jdddd}||fS)NtaskrF sqs_message AttributesApproximateReceiveCount)rEheadersr% properties)r=r>r@message_headersrVrWs rrRz+QoS.extract_task_name_and_number_of_retriess`/,/!/#F+    / >+ --F HII+++r F) r,r-r.r/r:r;r<rR __classcell__rCs@rr6r6sy??JJJJJJ C C C ,,,,,,,r r6ceZdZdZdZdZdZdZdZiZ dZ iZ iZ e ZeZfdZdZd Zfd Zfd Zd:d Zd ZefdZdZdZdZfdZdZedZdZ dZ!e"dfdZ#dZ$d;dZ%dZ&e"fdZ'e"dfdZ(dfd$ Z-d%Z.d&Z/fd'Z0 d;d(Z1d;d)Z2d*Z3d+Z4d;d,Z5e6d-Z7e6d.Z8e9d/Z:e9d0Z;e9d1Ze9d4Z?e9d5Z@e9d6ZAe9d7ZBe9d8ZCe9d9ZDxZES)?Channelz SQS Channel.z us-east-1ir#zkombu%(vhost)sNcttdtj|i||||j|dp t|_ dS)Nzboto3 is not installedhub) r ImportErrorr9__init___validate_predifined_queues_update_queue_cachequeue_name_prefixrJr rg)r=argskwargsrCs rrizChannel.__init__s =677 7$)&))) ((***   !7888::e$$8(8(8r cV|jD]\}}|dd}|d}|r*|s(td|d|s+|r)td||ddS)zCheck 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)rIitemsendswithr2format)r= queue_nameqfifo_url fifo_names rrjz#Channel._validate_predifined_queuess "399;;  MJx((11H"++G44I  +((.qx(8(8 ) +!6*ah77  r cB|jr1|jD]\}}|d|j|<dS||}|dgD]'}|dd}||j|<(dS)Nrp)QueueNamePrefix QueueUrls/)rIrrrQrS list_queuesrJsplit)r=rlrurvresprps rrkzChannel._update_queue_cache's  ! !%!7!=!=!?!? 9 9 A01%!*-- Fxxzz%%6G%HH88K,, 0 0C3+J,/D j ) ) 0 0r c|r|j||jr||t j||g|Ri|SN) _noack_queuesaddrg_loop1r9 basic_consume)r=queueno_ackrmrnrCs rrzChannel.basic_consume2sx  *   " "5 ) ) ) 8  KK   $uww$ 6     $*   r c||jvr'|j|}|j|t |Sr) _consumers _tag_to_queuerdiscardr9 basic_cancel)r= consumer_tagrrCs rrzChannel.basic_cancel;sL 4? * *&|4E   & &u - - -ww##L111r c |jr|jst||j||dS)zReturn a single payload message from one of our queues. Raises ------ Queue.Empty: if no messages available. )timeoutN)rqos can_consumer_pollcycle)r=rcallbackrns r drain_eventszChannel.drain_eventsAsN dh&:&:&<&< ''M 4:x 99999r cZtj|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)r FairCycle _get_bulk_active_queuesr_cycler=s r _reset_cyclezChannel._reset_cycleOs(!* ND/   r c"|drL|dtd }tt||}|dzStt||S)z3Format AMQP queue name into a legal SQS queue name.rqN)rslenstrr translate)r=nametablepartials r entity_namezChannel.entity_name]s| == ! ! 8>S\\M>*G(7++,,66u==GW$ $x~~&&0077 7r c<||j|zSr)rrl)r=rus rcanonical_queue_namezChannel.canonical_queue_namefs 6 CDDDr c ||}||jvr|| |j|S#t$r|jr"t d|dt|ji}| drd|d<| ||}|d|j|<|dcYSwxYw)zEnsure a queue with given name exists in SQS. Arguments: --------- queue (str): the AMQP queue name Returns str: the SQS queue URL rrMNr0)rrrsruuiduuid4sqs_base64_encodingrencoderrSrrJrT send_message)r=rr@rnq_urlbodyrs r_putz Channel._puts&&e$ 7 " "#w|'<<<L)--.BCC*+~~g&& >#w|'<<< -.>?+,,09F+,+w|/DDD -.FG3448;4:<<7H7HF344!W\%::: -n=>*  # ">>((w88DD>>D $} HH444U;;H < < ;;} % % % ' '%l3NC"# (      AN $ $V $ $ $ $ $r c tj|}tj||kr|Sn#t$rYnwxYw|Sr)base64 b64decode b64encode Exception) byte_stringdatas r_optional_b64_decodezChannel._optional_b64_decodes_ #K00D%%44 5    D s -1 >>c"||d}tt|}||jvrF||}||||dnv |d}|dd}n;#t$r.i}d|i}| t||dYnwxYw| ||d|d|d<|S) NBodyrrOr_rF)rr_r[ sqs_queuer>) rrrrrrasynsqsdelete_messagerGr)r=r@rurrpayloadr_rFs r_message_to_pythonzChannel._message_to_pythonsP(()?)?)A)ABB T**++ + + +OOJ//E LLzL * * 9 9(     $\2 ' 5o F     " -}= (..",      &U""   *1)AJ~ &sB115C)(C)cRfd|DS)aConvert 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 c>g|]}|Sr0)r)rmrrr=s r z/Channel._messages_to_python..s+KKKQ''5%88KKKr )r)r=messagesrrs` `@r_messages_to_pythonzChannel._messages_to_pythons8 &&KKKKKK(KKKKr c|}|r||}|||||j}|drr|dD]-}t |d|d<.||d|D]}|j ||dSt)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] rrNMaxNumberOfMessagesWaitTimeSecondsMessagesrrN) _get_message_estimaterrSreceive_messagewait_time_secondsrJrdecoder connection_deliverr) r=rmax_if_unlimitedr max_countrrrmsgs rrzChannel._get_bulks<..00  OOE**E88%8((88I $ 6988Dxx ## j)FFA ,!F) < < < C C E EAfII33D4DeLL99CO,,S%8888gg r c||}|||d|j}|drgt |ddd}||ddd<||d|dSt)z/Try to retrieve a single message off ``queue``.rrrrrrr) rrSrrrJrrrr)r=rrrrs r_getz Channel._get4s&&xxex$$44 2544 88J   HT*%5a%8%@AAAHHJJD*.D Q  '++D,multipler@r[r exceptionrCs rrzChannel.basic_acksw 0hll<00>G!-0KE''11'-2HII 0u%%44$[1"-o">5!!,///// 3 3 3%g.v6.HH4!*73I>$$\222222222  3 , , , GG  l + + + + + + ,s*'D 7B&& D0ADD+EEc||}|||}||dg}t |ddS)z)Return the number of messages in a queue.rApproximateNumberOfMessages)rNAttributeNamesr\)rrSrget_queue_attributesr%)r=rrrrs r_sizez Channel._sizesr&& HH444U;;H < <%%9:&<<4 %&CDEEEr c||}d}tdD]+}|t||z }|sn,||||S)z'Delete all current messages in a queue.rr#r)rN)rranger%rrS purge_queue)r=rrsizeis r_purgezChannel._purges&&r  A C 5))** *D   u))5)999 r cHtdSr)r9close)r=rCs rr%z Channel.closes  r ctj||||}|j|jnd}d|i}|j |j|d<|jdpi}tdi|} |jd d| i|S) N) region_nameaws_access_key_idaws_secret_access_keyaws_session_tokenTuse_ssl endpoint_urlz client-configrSconfigr0)rS) r sessionSession is_securer,rrJrclient) r=region access_key_idsecret_access_key session_tokenr.r0 client_kwargs client_configr-s rnew_sqs_clientzChannel.new_sqs_clients-''+"3+ (   '+n&@DNNd y    (,0,=M. ).22?CCIr ((-((w~DDFDmDDDr c| |jr||jvrtd|d|j|}|jdr|||S|jds||jvr |j|S||d|j|d|jj |d|jj x}|j|<|S|j |j S||j|jj |jj x}|_ |S)NzQueue with name 'z)' must be defined in 'predefined_queues'. sts_role_arnr2r3r4)r2r3r4) rIr*rrJ_handle_sts_session_predefined_queue_clientsr8r2conninfouseridpassword_sqsr=rrvrs rrSz Channel.sqss  !7 D222-////000&u-A%)).99 ://q999)--n== D:::9%@@++#$554;#?#?*+%% /1E+G+G./ee 3T]5K/M/M ,A6u=H 9 9 ++;-."m4,   DI r c,t|ds||jd|jdd}|d|_||d|j|d|d|d  x}|j|<|S|jd tj kr||jd|jdd}|d|_||d|j|d|d|d  x}|j|<|S|j|S) Nsts_expirationr:sts_token_timeouti Expirationr2 AccessKeyIdSecretAccessKey SessionToken)r2r3r4r5)tzinfo) hasattrgenerate_sts_session_tokenrrJrCr8r2r<replacerutcnow)r=rrv sts_credsrs rr;zChannel._handle_sts_sessionst-.. 977&**>::&**+>DDFFI#,L"9D 8<8K8KuuXt{33' 6"+,=">'7 9L99 A.u5 H  ( ( ( 5 58I8I I I77&**>::&**+>DDFFI#,L"9D 8<8K8KuuXt{33' 6"+,=">'7 9L99 A.u5 H1%8 8r cjtjd}||d|}|dS)NstsCelery)RoleArnRoleSessionNameDurationSeconds Credentials)r r1 assume_role)r=role_arntoken_expiry_seconds sts_client sts_policys rrKz"Channel.generate_sts_session_token sA\%(( ++$0,  -((r c||jr||jvrt|ds |j|S||jvr"td||j|}t |||d|jx}|j|<|S|j |j St |||jx}|_ |S)NrCrrr2)sqs_connectionr2) rI_predefined_queue_async_clientsrJr*rtrrSrJr2_asynsqsrAs rrzChannel.asynsqss  !7 <<<4!122=;EBBD222-+&--!!!&u-A"#'88%8#8#8554;77 A4U; H = $= .88%800;    DMr c|jjSr)rr1rs rr=zChannel.conninfo/s %%r c$|jjjSr)rr1rrs rrzChannel.transport_options3s%77r cD|jdp|jS)Nr)rrJdefault_visibility_timeoutrs rrzChannel.visibility_timeout7s'&**+?@@0/ 1r c8|jdiS)z/Map of queue_name to predefined queue settings.rIrrJrs rrIzChannel.predefined_queues<s%))*=rBBBr c8|jddS)Nrlrdrs rrlzChannel.queue_name_prefixAs%))*=rBBBr cdS)NFr0rs rsupports_fanoutzChannel.supports_fanoutEsur ct|jdptjjp|jS)Nr2)rrJr r/r'default_regionrs rr2zChannel.regionIs7&**844$ +$# %r c6|jdS)N regioninfordrs rrlzChannel.regioninfoOs%)),777r c6|jdS)Nr0rdrs rr0zChannel.is_secureSs%))+666r c6|jdSNportrdrs rrpz Channel.portWs%))&111r c|jjJ|jrdnd}|jjd|jj}nd}d||jj|SdS)Nhttpshttp:rfz {}://{}{})r=hostnamer0rprt)r=schemerps rr,zChannel.endpoint_url[so = ! - $:WWFF}!-/4=-//%% &  . -r cB|jd|jS)Nr)rrJdefault_wait_time_secondsrs rrzChannel.wait_time_secondsis)%))*=*.*HJJ Jr c8|jddS)NrTrdrs rrzChannel.sqs_base64_encodingns%))*?FFFr )NNr)rN)rra)Fr,r-r.r/rjrbrx domain_formatr^r]r@r<rQsetrr6rirjrkrrrrCHARS_REPLACE_TABLErrrrrr staticmethodrrrSQS_MAX_MESSAGESrrrrrrrrrr rrr#r%r8rSr;rKrpropertyr=rrrrIrlrhr2rlr0rpr,rrrbrcs@rreres N!% "$MH&(# D "LCEEM C 9 9 9 9 9( 0 0 0     22222 : : : :    ':8888EEE"$"$"$H   +++++'%'%'%R  \ 8LLL($4d****X   8888###6F    *:D     ---59     )E))))))0000002FFF   9=EEEE$    D999:)))6&&X&88X811_1CC_CCC_C_%%_% 88_877_722_2  _ JJ_JGG_GGGGGr receZdZdZeZdZdZdZej j e j e jfzZ ej je j fzZdZdZej jdedgZed 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 rrNrSTdirect) asynchronous exchange_typecd|jiSro) default_portrs rdefault_connection_paramsz#Transport.default_connection_paramss)**r )r,r-r.r/repolling_intervalrrrrconnection_errorsr BotoCoreErrorsocketerrorchannel_errors driver_type driver_name implementsextend frozensetrrr0r rrrss:GL+  !6<0 1 (J,D+FFKK"-44i ++5J ++X+++r r)6r/ __future__rrrstringrrrrbotocore.clientrbotocore.exceptionsrvinerr r kombu.asynchronousr kombu.asynchronous.aws.extr r %kombu.asynchronous.aws.sqs.connectionr"kombu.asynchronous.aws.sqs.messager kombu.logr kombu.utilsrkombu.utils.encodingrrkombu.utils.jsonrrkombu.utils.objectsrrfrr,logger punctuationr|r~r(rr*r2r4r6rerr0r rrsAAH#""""" """"""++++++3333333333------88888888DDDDDD;;;;;; """"""77777777))))))))////// H   ,!DLLLLLiLLLKKKKKIKKK0,0,0,0,0,'+0,0,0,fz Gz Gz Gz Gz Ggoz Gz Gz Gz4+4+4+4+4+!4+4+4+4+4+r