> i׭|dZddlmZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z ddl m Z ddlmZddlmZddlmZdd lmZdd lmZmZmZdd lmZdd lmZmZdd lm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(m)Z)ddl*m+Z+m,Z,ddl-m.Z.ddl/m0Z0e$e1Z2ejfDcic]}|dvse4|dc}Z5de5d<dZ6dZ7Gdde8Z9Gdde8Z:Gd d!e8Z;Gd"d#e8Z<Gd$d%e0jzZ=Gd&d'e0j|Z>Gd(d)e0j~Z?ycc}w)*arAmazon 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 'sts_token_buffer_time': 0, # optional, added in 5.6.0 } 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. .. versionadded:: 5.6.0 sts_token_buffer_time (seconds) is the time by which you want to refresh your token earlier than its actual expiration time, defaults to 0 (no time buffer will be added), should be less than sts_token_timeout. 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 timedeltatimezone)JSONDecodeError)Empty)Any)Config) ClientError)ensure_promisepromise transform)get_event_loop)boto3 exceptions)AsyncSQSConnection) AsyncMessage) get_logger) scheduling) bytes_to_strsafe_str)dumpsloads)cached_property)virtualz-_._-. c< t|S#t$r|cYSwxYw)z5Try to convert x' to int, or return x' if that fails.)int ValueError)xs o/var/www/html/wine-match-dev/backend/winematch-backend/venv/lib/python3.12/site-packages/kombu/transport/SQS.py maybe_intr&s$1v s ceZdZdZy)UndefinedQueueExceptionzAPredefined queues are being used and an undefined queue was used.N__name__ __module__ __qualname____doc__r%r(r(sKr/r(ceZdZdZy)InvalidQueueExceptionz@Predefined queues are being used and configuration is not valid.Nr)r.r/r%r1r1sJr/r1ceZdZdZy)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)r.r/r%r3r3sr/r3ceZdZdZy)DoesNotExistQueueExceptionz"The specified queue doesn't exist.Nr)r.r/r%r5r5s,r/r5c6eZdZdZdfd ZdZdZdZxZS)QoSz5Quality of Service guarantees implementation for SQS.ct||||j|\}}}}|r|r|r|r|j||||yyyyy)N)requeue)superreject1_extract_backoff_policy_configuration_and_messageapply_backoff_policy)self delivery_tagr9 routing_keymessage backoff_tasksbackoff_policy __class__s r%r;z QoS.rejects` |W5  B B  < Wm^ 7}  % %\>= J:H}7;r/c |j|}|jd}|r|sy|jjj |i}|j d}|j d}||||fS#t$rYywxYw)Nr@)NNNNrBrC) _delivered delivery_infoKeyErrorchannelpredefined_queuesget)r>r?rAr@ queue_configrBrCs r%r<z5QoS._extract_backoff_policy_configuration_and_messages *ool3G!// >K')||5599+rJ $((9 %))*:;G]NBB *) *sA33 A?>A?c|jj|}|j|\}}|r|sy|j|}||vr3|0|jj |} | j |||yyy)NQueueUrl ReceiptHandleVisibilityTimeout)rI _queue_cache'extract_task_name_and_number_of_retriesrKsqschange_message_visibility) r>r@r?rCrB queue_url task_namenumber_of_retries policy_valuecs r%r=zQoS.apply_backoff_policysLL--k:  8 8 F % $ 1%))*;<  %,*B   -A ' '"*". ( +C %r/c|j|}|j}|d}t|jdddd}||fS)NtaskrG sqs_message AttributesApproximateReceiveCount)rFheadersr" properties)r>r?rAmessage_headersrWrXs r%rSz+QoS.extract_task_name_and_number_of_retriessa//,/!//#F+     / >+ --F HI+++r/F) r*r+r,r-r;r<r=rS __classcell__rDs@r%r7r7s?J C ,r/r7ceZdZdZdZdZdZdZdZiZ dZ iZ iZ e ZeZej"dZfdZd Zd Zfd Zfd ZdBd ZdZefdZdZdZdZdZdZ dZ!dZ"dZ# dC dDdZ$e%dfdZ&dZ'dEdZ(dZ)e%fdZ*dEdZ+dCdZ,d Z- dCd!Z. dFfd" Z/dGfd# Z0d$Z1d%Z2fd&Z3 dEd'Z4dEd(Z5d)Z6dHd*Z7d+Z8d,Z9dEd-Z:e;d.Zd0Z?e>d1Z@e>d2ZAe>d3ZBe>d4ZCe>d5ZDe>d6ZEe>d7ZFe>d8ZGe>dId9ZHe>d:ZIe>d;ZJe;dJd<ZKdKd=ZLd>ZMd?ZNd@ZOdAZPxZQS)LChannelz SQS Channel.z us-east-1ir zkombu%(vhost)sNs@^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$ct tdt| |i||j |j |j |jdxs t|_ y)Nzboto3 is not installedhub) r ImportErrorr:__init___validate_predifined_queues_update_queue_cachequeue_name_prefixrKrri)r>argskwargsrDs r%rkzChannel.__init__s` =67 7 $)&) ((*   !7!78::e$8(8r/c|jjD]n\}}|djd}|jd}|r|stdj |d|rO|sRtdj ||dy)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)rJitemsendswithr1format)r> queue_nameqfifo_url fifo_names r%rlz#Channel._validate_predifined_queues#s "3399; MJx((1H"++G4I +((.qx(8)+!6*ah7 r/c:|jr5|jjD]\}}|d|j|<y|jj |}|j dgD]%}|j dd}||j|<'y)Nrr)QueueNamePrefix QueueUrls/)rJrtrRrT list_queuesrKsplit)r>rnrwrxresprrs r%rmzChannel._update_queue_cache7s  ! !!%!7!7!=!=!? 9 A01%!!*- 9 xxz%%6G%H88K, 0C3+J,/D  j ) 0r/c|r|jj||jr|j|t |||g|i|SN) _noack_queuesaddri_loop1r: basic_consume)r>queueno_ackrorprDs r%rzChannel.basic_consumeBsT     " "5 ) 88 KK w$ 6   $*  r/c||jvr*|j|}|jj|t||Sr) _consumers _tag_to_queuerdiscardr: basic_cancel)r> consumer_tagrrDs r%rzChannel.basic_cancelKsC 4?? *&&|4E    & &u -w#L11r/c |jr|jjs t|j |j ||y)zReturn a single payload message from one of our queues. Raises ------ Queue.Empty: if no messages available. )timeoutN)rqos can_consumer_pollcycle)r>rcallbackrps r% drain_eventszChannel.drain_eventsQs9dhh&:&:&<'M 4::x 9r/cltj|j|jt|_y)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_cycle_s'!** NND//  r/c|jdr7|dtd }tt|j |}|dzStt|j |S)z3Format AMQP queue name into a legal SQS queue name.rsN)rulenstrr translate)r>nametablepartials r% entity_namezChannel.entity_namems_ == !>S\M*G(7+,66u=GW$ $x~&007 7r/c>|j|j|zSr)rrn)r>rws r%canonical_queue_namezChannel.canonical_queue_namevs 6 6 CDDr/c|j|}||jvr|j| |j|S#t$r6|jrt dj |td|dwxYw)z9Try to retrieve the SQS queue URL for a given queue name.r sqs_qnames r%_resolve_queue_urlzChannel._resolve_queue_urlys--e4 D-- -  $ $Y / $$Y/ / %%-6&#%% -#I;.DE  s A?Bc  |j|S#t$rj|j|}dt|ji}|j drd|d<|j ||}|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 rQrstrue FifoQueuerO)rr5rrvisibility_timeoutru _create_queuerR)r>rrpr attributesrs r% _new_queuezChannel._new_queues $**51 1) $11%8I-s43J3J/KLJ!!'**0 ;'%%irwrr create_paramss r%rzChannel._create_queues  ! !  " " & &'@ A GR ++// = $$ $.M& !6txxjx)66GGGr/c|jry|j|}|jj||jj |dy)zDelete queue by name.NrO)rJrrT delete_queuerRpop)r>rrorpq_urls r%_deletezChannel._deletesR  ! ! ''.     eT*r/c |j|}d|i}d|vrd|dvr|djd|d<|jdrLd|dvr |dd|d<nd|d<d|dvr |dd|d<n3tt j |d<nd |dvr |dd |d <|j r#tjt|}n t|}||d <|j|j| }|jd r%|j||dd |jy|jdi|y)zPut message onto queue.rOramessage_attributesMessageAttributesrsMessageGroupIddefaultMessageDeduplicationId DelaySeconds MessageBodyr redeliveredr?rNNr.)rrruruuiduuid4sqs_base64_encodingrencoderrTrrKrUwait_time_seconds send_message)r>rrArprbodyrZs r%_putz Channel._puts&e$ 7 "#w|'<<L)--.BC*+~~g&#w|'<< -.>?+,09F+,+w|/DD -.FG348;4::<7HF34!W\%:: -n=>*  # #>((w8D>D $} HH444U;H < ;;} % ' '%l3NC"&"8"8 (  ANN $V $r/c|d}|j|}t|}|j|}||jvr|j |||S|j ||||S)NBody)_decode_python_message_bodyr_prepare_json_payloadr_delete_message_envelope_payload)r>rArwr raw_msg_body decoded_bytestextpayloads r%_message_to_pythonzChannel._message_to_pythonsqv 88F M*,,T2 ++ +  W 5N%%gtWeDDr/cp|j|}|Dcgc]}|j|||c}Scc}w)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 )rr)r>messagesrrms r%_messages_to_pythonzChannel._messages_to_pythons5 &BJKQ''5%8KKKs3c|j|}|j|}|jjd}|jjd}|||xs |j||d}|j di|S)aMUnified receive_message wrapper for SQS (boto3.client.SQS) with full attribute support. :param queue: The queue as a string :param max_number_of_messages: Int of max number of messages to receive. :param wait_time_seconds: Int of sqs wait time in seconds. :return: SQS client recieve_message rMessageSystemAttributeNamesMessageAttributeNames)rOMaxNumberOfMessagesWaitTimeSecondsrrr.)rrTget_message_attributesrKrreceive_message) r>rmax_number_of_messagesrrclientmessage_system_attribute_namesmessage_attribute_namesparamss r%_receive_messagezChannel._receive_messages__U+&)-)D)D)H)H )*+&"&"="="A"A ##%#90JD4J4J%<+I " &v%%///r/cZ|j}|r|j||j|}|jdra|dD]"}t |dj |d<$|j |d|D]}|jj|| yt)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] rrrMessagesrrN) _get_message_estimaterrrKrdecoder connection_deliverr)r>rmax_if_unlimitedr max_countrrmsgs r%rzChannel._get_bulk2s<..0 (("&"8"8'0)D xx #j)FA ,!F) < C C EAfIF33D4DeL9COO,,S%89g r/c|j||jd}|jdrFt|dddj }||ddd<|j |d|dSt )z/Try to retrieve a single message off ``queue``.rrrrrr)rrrKrrrr)r>rrrs r%_getz Channel._get`s$$"44#$%  88J T*%5a%8%@AHHJD*.D Q  '++D,r_s r%rzChannel._loop1ns 4//7r/c||jvrU|jjr)|j|t |j |fy|j |yyN)r)rrr_get_bulk_asyncr r)r>rs r%rzChannel._schedule_queueqsX D'' 'xx##%$$GDKK%$B% E" (r/ct|jj}t|||St|d|S)Nr)rcan_consume_max_estimateminmax)r>rmaxcounts r%rzChannel._get_message_estimatezsD88446 ( 0    69(A6F   r/cx|j}|r|j|||St|}|g|Sr)r _get_asyncr )r>rrr s r%rzChannel._get_bulk_asyncs>--/ ??5(X?F F!(+ r/c |j|}|j|}|j||||j|t |j |||S)Nr)rwrVcountrr)rr _get_from_sqsasynsqsr_on_messages_ready)r>rrrrqnames r%r zChannel._get_asyncsd&))%0!!U||%|0''5%"  r/cd|vrE|dr?|jj}|dD] }|j|||}|||"yyy)Nr)r _callbacksr)r>rrr callbacksr msg_parseds r%rzChannel._on_messages_readys_  !hz&:22I + -!44S%G  % , -'; !r/cB|j||||j|S)zwRetrieve and handle messages from SQS. Uses long polling and returns :class:`~vine.promises.promise`. )number_messagesrr)rr)r>rwrVrrrs r%rzChannel._get_from_sqss/ ))  5"44*  r/ch|D]}|jj|d t| |Sr)rGrr:_restore)r>rAunwanted_delivery_info unwanted_keyrDs r%rzChannel._restores:2 :L  ! ! % %lD 9 :w((r/c |jj|j}|d}d}d|vr|j|d} |j |j |d|dt ||y#t$rI}|jddd k(rt|jdd t |-|Yd}~yd}~wwxYw#t$rt ||YywxYw) Nr]r@r sqs_queuerP)rOrPErrorCode AccessDeniedMessage) rrKrGrrTdelete_messager: basic_ackr responser3 basic_rejectrH)r>r?multiplerAr]r exceptionrDs r%r$zChannel.basic_acks  0hhll<0>>G!-0KE'11'-2HI 0u%44$[1"-o">5!,/ 3%%g.v6.H4!**73I>$\22  3 , G l + ,s)*C)B C ?CCC10C1c|j|}|j|j|}|j|dg}t |ddS)z)Return the number of messages in a queue.rApproximateNumberOfMessages)rOAttributeNamesr^)rrTrget_queue_attributesr")r>rrrZrs r%_sizez Channel._sizesa& HH444U;H <%%9:&<4 %&CDEEr/c|j|}d}tdD]#}|t|j|z }|r#n|j |j ||S)z'Delete all current messages in a queue.rr rr)rranger"r-rT purge_queue)r>rrsizeis r%_purgezChannel._purgesk&r A C 5)* *D  u))5)9 r/c"t|yr)r:close)r>rDs r%r5z Channel.closes   r/c6tjj||||}|j |jnd}d|i}|j|j|d<|j j dxsi}tdi|} |jdd| i|S) N) region_nameaws_access_key_idaws_secret_access_keyaws_session_tokenTuse_ssl endpoint_urlz client-configconfigr.)rT) rsessionSession is_securer<rrKr r) r>region access_key_idsecret_access_key session_tokenr>r@ client_kwargs client_configr=s r%new_sqs_clientzChannel.new_sqs_clients--''+"3+ (  '+nn&@DNNd y     (,0,=,=M. )..22?CIr (-(w~~DFDmDDr/c |%|jr||jvrtd|d|j|}|jjdr|j ||S|jjds||j vr|j |S|j |jd|j|jd|jj|jd|jjx}|j |<|S|j |jS|j |j|jj|jjx}|_ |S)Nrz)' must be defined in 'predefined_queues'. sts_role_arnrArBrC)rArBrC) rJr(rrK_handle_sts_session_predefined_queue_clientsrGrAconninfouseridpassword_sqs)r>rrxrZs r%rTz Channel.sqss|  !7!7D222-'w///00&&u-A%%)).9//q99))--n=D:::99%@@++#$554;;#?*+%% /1E1E+G./ee 3T]]5K5K/M ,A66u=H 99 99 ++;;--.."mm44,  DI r/c|jd|j}t|ds|j||S|jj dt jtjj dkr|j||S||jvr|j||S|j|S)NrAsts_expiration)tzinfo) rKrAhasattr-_new_predefined_queue_client_with_sts_sessionrQreplacernowrutcrK)r>rrxrAs r%rJzChannel._handle_sts_sessionsx-t-.EEeVT T  ( ( ( 5 X\\8R8Z8Zbf8Z8g gEEeVT TD:::II%QWXX11%8 8r/cz|j||}|r&d|cxkr|krn|S|dxxt|zcc<|S)zGenerate STS session credentials with an optional expiration buffer. The buffer is only applied if it is less than `token_expiry_seconds` to prevent an expired token. r Expiration)seconds)generate_sts_session_tokenr)r>role_arntoken_expiry_secondstoken_buffer_seconds credentialss r%&generate_sts_session_token_with_bufferz.Channel.generate_sts_session_token_with_buffer'sL 55h@TU A(<$S?S$S  %;O)P P %r/c4|j|jjd|jjdd|jjdd}|d|_|j ||d|d|d  x}|j |<|S) NrIsts_token_timeoutists_token_buffer_timerrY AccessKeyIdSecretAccessKey SessionToken)rArBrCrD)r`rrKrQrGrK)r>rrA sts_credsrZs r%rTz5Channel._new_predefined_queue_client_with_sts_session1s??  " " & &~ 6  " " & &':C @  " " & &'> B ( 5484G4G#M2'(9:#N3 5H5  D * *5 1 r/c^tjd}|j|d|}|dS)NstsCelery)RoleArnRoleSessionNameDurationSeconds Credentials)rr assume_role)r>r\r] sts_client sts_policys r%r[z"Channel.generate_sts_session_token@s;\\%( ++$0, -((r/cr|jjd}|jjd}||jr||jvrt |ds|j|S||jvrt dj ||j|}t|j||jd|j||x}|j|<|S|j |jSt|j||j||x}|_ |S)NrrrQrrrA)sqs_connectionrArr) rrKrJ_predefined_queue_async_clientsrSr(rvrrTrA_asynsqs)r>rrrrxrZs r%rzChannel.asynsqsIs=)-)D)D)H)H )*+&"&"="="A"A ##%  !7!7<<<4!12;;EBBD222-+&-!!&&u-A"#'88%8#8554;;73Q,C  A44U;H == $== .88%80;;+I$;   DM r/c.|jjSr)rrrs r%rLzChannel.conninfoms%%%r/cB|jjjSr)rrrrs r%rzChannel.transport_optionsqs%%777r/cT|jjdxs |jS)Nr)rrKdefault_visibility_timeoutrs r%rzChannel.visibility_timeoutus*&&**+?@0// 1r/c:|jjdiS)z/Map of queue_name to predefined queue settings.rJrrKrs r%rJzChannel.predefined_queueszs%%))*=rBBr/c:|jjddS)Nrnr{rs r%rnzChannel.queue_name_prefixs%%))*=rBBr/cy)NFr.rs r%supports_fanoutzChannel.supports_fanoutsr/c|jjdxs,tjjxs |j S)NrA)rrKrr?r7default_regionrs r%rAzChannel.regions=&&**84$ ++$## %r/c8|jjdS)N regioninfor{rs r%rzChannel.regioninfos%%)),77r/c8|jjdS)Nr@r{rs r%r@zChannel.is_secures%%))+66r/c8|jjdSNportr{rs r%rz Channel.ports%%))&11r/c|jji|jrdnd}|jjd|jj}nd}dj ||jj|Sy)Nhttpshttp:r}z {}://{}{})rLhostnamer@rrv)r>schemers r%r<zChannel.endpoint_urlsv == ! ! - $WFF}}!!-4==--./%% &&  .r/cN|jjd|jS)Nr)rrKdefault_wait_time_secondsrs r%rzChannel.wait_time_secondss)%%))*=*.*H*HJ Jr/c:|jjddS)NrTr{rs r%rzChannel.sqs_base64_encodings%%))*?FFr/c:|jjddS)Nfetch_message_attributesr{rs r%rz Channel.fetch_message_attributess%%))*DdKKr/c4d}|j}d}d}|t|trd|gdSt|trJdj |Dcgc]}|j c}vrdgntt ||gz}nt|t r|jdg}|jdd}t|trIdj |Dcgc]}|j c}vrdgntt ||gz}t|trG|rEdj |Dcgc]}|j c}vrdgntt |}|r t|nd|r t|dS|gdScc}wcc}wcc}w)aGet the message attributes to be fetched from SQS. Ensures 'ApproximateReceiveCount' is included in system attributes if list is provided. - The number of retries is managed by SQS / (specifically by the ``ApproximateReceiveCount`` message attribute) - See: class QoS(virtual.QoS): (method) def extract_task_name_and_number_of_retries :return: A dictionary with SQS message attribute fetch config. r_N)rrALLrr) r isinstancerlistlowersetdictrKsorted)r>APPROXIMATE_RECEIVE_COUNTfetchmessage_system_attrs message_attrsssystemattrss r%rzChannel.get_message_attributess%>!--# =Juc2)-0I/J  eT ".3kkmSX?Ya ?Y.YE7S";!<<=> !t $YYraw candidatedecoded reencodeds r%_optional_b64_decodezChannel._optional_b64_decodes IIK >> # #I . 6J &&y4@G$$W-44T:  ((. .J  NN7 # + J " J s#BB5B21B25 CCcht|tr|jn|}|j|S)zDecode the message body when needed. raw_body: bytes or str returns: bytes (decoded Base64 if it looks like Base64, otherwise raw bytes) )rrrr)r>raw_bodybs r%rz#Channel._decode_python_message_bodys, ",Hc!:HOO ((++r/cp t|}t|tr|SiS#ttf$ricYSwxYw)z:Try to JSON-decode text into a dict; on failure return {}.)rrrr TypeError)r>rdatas r%rzChannel._prepare_json_payload s= ;D%dD14 9r 9+ I s!!55cp|j|}|j|j||dy)z9Move the message over to the new queue URL and delete it.rrPN)rrr#)r>rwrAnew_qs r%rzChannel._delete_messages3 + : &55 7?+ r/cd|vr|diid}|jdi}|jdi}|j||d|d|d<|S)aPrepare the payload envelope. Ensure we have a dict with 'body' and 'properties.delivery_info', then stamp on SQS-specific metadata. :param payload: The payload as an object :param raw_text: Text that will be set as the payload body. :param message: A kombu Message. :param q_url: The SQS queue URL. :return: Payload object. rarG)rrar]rrPr?) setdefaultr)r>rraw_textrArpropsdis r%rzChannel._envelope_payloadss w & .3G ""<4   or 2 "   !( 8nr/)NN)rN)rrrr"rz int | Noner)rrc)r)returnr")rzdict[str, Any])rbytesrr)Rr*r+r,r-rryr domain_formatrurtrOrKrRrrr7recompilerrkrlrmrrrrCHARS_REPLACE_TABLErrrrrrrrrrSQS_MAX_MESSAGESrrrrrrr rrrr$r-r3r5rGrTrJr`rTr[rpropertyrLrrrrJrnrrArr@rr<rrrrrrrrrrdres@r%rgrgs N!% "$MH&(# D "LEM C _`I 9( 0 2 :  ':8E2$*H. +'%R EL,'((, 00!$0& 0@$4d,\ 8#6F   -59  )E)02F 9=E$ D 9 )"H&&8811CCCC%% 887722  JJGGLL. . h6, r/rgc:eZdZdZeZdZdZdZejje je jfzZ ejje jfzZdZdZejj$j'dedgZed Zy) TransportaJ SQS 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 .. versionadded:: 5.6 Queue tags can be applied to SQS queues during creation by passing an ``queue_tags`` key in transport_options. ``queue_tags`` must be a dict of tag key-value pairs. .. code-block:: python from kombu.transport.SQS import Transport transport = Transport( ..., transport_options={ 'queue_tags': { 'Environment': 'production', 'Team': 'backend', }, } ) 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"], # Get all of the MessageSystemAttributeNames (formerly AttributeNames) } ) # Preferred - A dict specifying system and custom message attributes transport = Transport( ..., transport_options={ 'fetch_message_attributes': { 'MessageSystemAttributeNames': ["SenderId", "SentTimestamp"], 'MessageAttributeNames': ['S3MessageBodyKey'] }, } ) .. _Message Attributes: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html#SQS-ReceiveMessage-request-AttributeNames rrNrTTdirect) asynchronous exchange_typecd|jiSr) default_portrs r%default_connection_paramsz#Transport.default_connection_paramss))**r/)r*r+r,r-rgpolling_intervalrrrrconnection_errorsr BotoCoreErrorsocketerrorchannel_errors driver_type driver_name implementsextend frozensetrrr.r/r%rr=sIVGL++  ! !6<<0 1 ((J,D,D+FFKK""--44 +5J ++r/r)@r- __future__rrrrrstringrrrrjsonrrrtypingr botocore.clientr botocore.exceptionsr viner r rkombu.asynchronousrkombu.asynchronous.aws.extrr%kombu.asynchronous.aws.sqs.connectionr"kombu.asynchronous.aws.sqs.messager kombu.logr kombu.utilsrkombu.utils.encodingrrkombu.utils.jsonrrkombu.utils.objectsrr}rr*logger punctuationordrrr& Exceptionr(r1r3r5r7rgr)rZs0r%rsGT#  22 "+33-8D; "7)/ H  !,,CFDL!DLiLKIK--0,'++0,fv goov rb+!!b+cs + D95 D9