O'^c@@sdZddlmZddlZddlZddlZddlmZmZddl Z ddl m Z ddl m Z ddl m ZddlmZdd lmZdd lmZdd lmZdd lmZmZmZdd lmZddlmZmZddl m!Z!m"Z"ddl#m$Z$ddl%m&Z&ee'Z(e)dej*DZ+de+d 1. Amazon supports up to 20 seconds. This is disabled for now, but will be enabled by default in the near future. Batch API Actions: http://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. i(tabsolute_importN(tloadstdumps(t exception(tsdb(tsqs(tDomain(t SDBConnection(t SQSConnection(tMessage(tEmptytrangettext_t(t get_logger(tcached_propertytuuid(t bytes_to_strtsafe_str(t schedulingi(tvirtualcc@s-|]#}|dkrt|dfVqdS(s-_.i_N(tord(t.0tc((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pys Esi-i.cC@s'yt|SWntk r"|SXdS(N(tintt ValueError(tx((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt maybe_intJs cc@s|]}t|VqdS(N(R(Rtpart((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pys Ost.iii tTablecB@seZdZeZdZdZdZdZdZ dZ dZ dde dd Zdd Zd Zd Zd ZRS(s<Amazon SimpleDB domain describing the message routing table.cC@s|jd|S(s+Iterator giving all routes for an exchange.sWHERE exchange = '%s'(tselect(tselftexchange((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt routes_for[scC@s&|j|}|r"|j|SdS(sGet binding for queue.N(t _get_queue_idtget_item(Rtqueuetqid((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt get_queue_scC@s?|j|}|r#||dfSt}|j||fS(sTGet binding item for queue. Creates the item if it doesn't exist. tid(R&Rtnew_item(RR$titemR'((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pytcreate_bindinges  c C@s||jkr{|j|\}}|jd|d|p<dd|pHdd|pTdd||j|jj|ndS(NR t routing_keyttpatternR$R'(t_already_boundR*tupdatetsavetadd(RR R+R-R$tbindingR'((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt queue_bindqs    cC@s9|jj||j|}|r5|j|ndS(sdelete queue by name.N(R.tdiscardt_get_queue_itemt delete_item(RR$R)((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt queue_delete|scC@s/x(|j|D]}|j|dqWdS(s!Delete all routes for `exchange`.R'N(R!R6(RR R)((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pytexchange_deletescC@s:x3ttfD]%}tj|||}|r |Sq WdS(s"Uses `consistent_read` by default.N(tFalsetTrueRR#(Rt item_nametconsistent_readR)((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR#sR,cC@s,d|j|f}tj|||||S(s"Uses `consistent_read` by default.sSELECT * FROM `%s` %s(tnameRR(Rtqueryt next_tokenR<t max_items((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscK@s?x8ttfD]*}x!|j|d||D]}|SWq WdS(NR<(R9R:R(RR>tkwargsRR)((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt _try_firstscC@s ttd|jDS(Ncs@s|]}|dVqdS(R N((Rti((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pys s(tlisttsetR(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt get_exchangesscC@s|jd|S(NsWHERE queue = '%s' limit 1(RB(RR$((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR5scC@s!|j|}|r|dSdS(NR'(R5(RR$R)((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR"sN(t__name__t __module__t__doc__RER.R!R&R*R3R7R8R#tNoneR:RRBRFR5R"(((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRVs         tChannelcB@seZeZdZdZdZdZd.Zd.Z iZ e Z dZ dZdZd.dZdZed Zd Zd.d d.d Zd ZdZdZdZdZdZdZdZddZdZdZ e!dZ"dZ#d/dZ$dZ%dZ&d Z'd!Z(d"Z)d#Z*e+d$Z,e+d%Z-e+d&Z.e+d'Z/e+d(Z0e1d)Z2e1d*Z3e1d+Z4e1d,Z5e1d-Z6RS(0s us-east-1iiskombu%(vhost)scO@sqtt|j|||jjd|j}x|D]}||j|jsN( R[tqost can_consumeR RUtpopleftt IndexErrort_polltcycletextend(RR_t message_cachetres((R$s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt drain_eventss    cC@s"tj|j|jt|_dS(sReset the consume cycle. :returns: a 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(Rt FairCyclet _get_bulkt_active_queuesR t_cycle(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt _reset_cycles cC@stt|j|S(s3Format AMQP queue name into a legal SQS queue name.(R Rt translate(RR=ttable((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt entity_namescK@sa|j|j|}y|j|SWn5tk r\|jj||j}|j|<|SXdS(s-Ensure a queue with given name exists in SQS.N(RrRPRQtKeyErrorRt create_queuetvisibility_timeout(RR$RAtq((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt _new_queues  R,cK@sQtt|j||||||j|jdkrM|jj|ndS(Ntfanout(RMRKR3ttypeofttypeRRR1(RR$R R+t argumentsRA((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR3 s cG@s |jr|jj|ndS(snBind ``queue`` to ``exchange`` with routing key. Route will be stored in SDB if so enabled. N(tsupports_fanoutRqR3(RRV((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt _queue_binds cC@sW|jrAg|jj|D]!}|d|d|df^qStt|j|S(sTGet routing table. Retrieved from SDB if :attr:`supports_fanout`. R+R-R$(R|RqR!RMRKt get_table(RR R`((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR~s 5cC@s)|jr|jjStt|jS(N(R|RqRFRMRK(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRF(s  cG@sI|jr|jj|ntt|j||jj|ddS(sdelete queue by name.N( R|RqR7RMRKt_deleteRQtpopRJ(RR$RV((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR-s cK@s9|jr|jj|ntt|j||dS(sDelete exchange by name.N(R|RqR8RMRK(RR RA((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR84s cK@s5|jrt|jj|Stt|j|S(s1Return True if ``queue`` was previously declared.(R|tboolRqR&RMRKt _has_queue(RR$RA((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR:s cK@s<|j|}t}|jt||j|dS(sPut message onto queue.N(RwR tset_bodyRtwrite(RR$tmessageRARvtm((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_put@s cK@s8x1|jj|D]}|j|d||qWdS(s5Deliver fanout message to all queues in ``exchange``.R$N(RqR!R(RR RR+RAtroute((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt _put_fanoutGsicC@sK|j|}tr:||jkr:|j|d|jS|j|SdS(s|Retrieve messages from SQS and returns the raw SQS message objects. :returns: List of SQS message objects twait_time_secondsN(RwtW_LONG_POLLINGRRt get_messagesR(RR$tcountRv((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt _get_from_sqsLs cC@s^tt|j}||jkr7|j|n#|ddji|d6|d6|S(Nt propertiest delivery_infot sqs_messaget sqs_queue(RRtget_bodyRXtdelete_messageR/(RRt queue_nameR$tpayload((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_message_to_pythonZs cC@s5|j|}g|D]}|j|||^qS(sConvert 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. :param messages: A list of SQS Message objects. :param queue: String name representing the queue they came from :returns: A list of Payload objects (RwR(RtmessagesR$RvR((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_messages_to_pythonds cC@sz|jj}|dkr!|n t|d}|rm|j|dt|t}|rm|j||SntdS(s+Try to retrieve multiple messages off ``queue``. Where _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 that 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. args: queue: The queue name (string) to pull from returns: payloads: A list of payload objects returned iRN( Ratcan_consume_max_estimateRJtmaxRtmintSQS_MAX_MESSAGESRR (RR$tmax_if_unlimitedtmaxcountR((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRlts!cC@s<|j|dd}|r/|j||dStdS(s/Try to retrieve a single message off ``queue``.RiiN(RRR (RR$R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_getsRRcC@s:x!|D]}|jj|dqWtt|j|S(N(RRRJRMRKt_restore(RRtunwanted_delivery_infot unwanted_key((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRs cC@sb|jj|j}y|d}Wntk r6nX|j|dtt|j|dS(NRR(RatgetRRsRRMRKt basic_ack(Rt delivery_tagRR$((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRs cC@s|j|jS(s)Return the number of messages in a queue.(RwR(RR$((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_sizescC@sT|j|}d}x.tdD] }||j7}|s"Pq"q"W|j|S(s'Delete all current messages in a queue.ii (RwR Rtclear(RR$RvtsizeRC((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_purges cC@sytt|jx_|j|jfD]K}|r&y|jWqqtk rm}dt|krnqnqqXq&q&WdS(Nscan't set attribute(RMRKtcloset_sqst_sdbtAttributeErrortstr(Rtconntexc((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscC@s7|jr3x'|D]}|j|jkr|SqWndS(N(tregionR=(Rtregionst_r((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_get_regioninfos  c C@s@|j}|j|}|d|d|jd|jd|jS(NRtaws_access_key_idtaws_secret_access_keytport(tconninfoRtuseridtpasswordR(RtfunRRR((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt_aws_connect_tos     cC@s4|jdkr-|jttj|_n|jS(N(RRJRRR(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscC@s4|jdkr-|jttj|_n|jS(N(RRJRRR(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscC@sR|j|ji|jjd6}|jjdi|d6|j}||_|S(Ntvhostt CreateDomaint DomainName(Rrt domain_formatRt virtual_hostRt get_objectRR=(RR=td((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRqs   cC@s |jjS(N(t connectiontclient(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscC@s |jjjS(N(RRttransport_options(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscC@s|jjdp|jS(NRu(RRtdefault_visibility_timeout(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRuscC@s|jjddS(NRPR,(RR(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRPscC@s|jjdtS(Ntsdb_persistence(RRR9(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR|scC@s|jjdp|jS(NR(RRtdefault_region(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRscC@s|jjd|jS(NR(RRtdefault_wait_time_seconds(R((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRs N(RR(7RGRHRRRRRRJRRRQRERXRNRYR]RjRotCHARS_REPLACE_TABLERrRwR3R}R~RFRR8RRRRRRRRlRRRRRRRRtpropertyRRRqRRRRuRPR|RR(((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyRKs`                 #     t TransportcB@s^eZeZdZdZdZejj e j e j fZ ejje jfZdZdZRS(iiRN(RGRHRKtpolling_intervalRRJt default_portRRtconnection_errorsRtSQSErrortsocketterrortchannel_errorstSQSDecodeErrort driver_typet driver_name(((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyR s (6RIt __future__RRSRtstringtanyjsonRRtbotoRRRRRtboto.sdb.domainRtboto.sdb.connectionRtboto.sqs.connectionRtboto.sqs.messageR t kombu.fiveR R R t kombu.logR t kombu.utilsRRtkombu.utils.encodingRRtkombu.transport.virtualRR,RRGtloggertdictt punctuationRRttuplet __version__tsplitt BOTO_VERSIONRRRRKR(((s5/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/SQS.pyt&s>        "Rf