U dǦ@sdZddlZddlZddlZddlmZddlmZmZddlm Z ddl m Z ddl m Z ddlmZmZmZdd lmZdd lmZmZdd lmZddlZdd lmZmZmZmZdd lm Z ddl!m"Z"ddlm#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*ddl+m,Z,m-Z-m.Z.m/Z/m0Z0ddl1m2Z2ddl3m4Z4m5Z5ddl6m7Z7ddl8m9Z9m:Z:m;Z;mm?Z?dZ@eAdhZBe7eCZDdZEeddZFdZGdZHddZIGdd d eJZKd!d"ZLGd#d$d$ZMGd%d&d&ZNGd'd(d(eMeNZOeOZPGd)d*d*eMZQGd+d,d,eQeNZRGd-d.d.eOZSdS)/zResult backend base classes. - :class:`BaseBackend` defines the interface. - :class:`KeyValueStoreBackend` is a common base class using K/V semantics like _get and _put. N) namedtuple)datetime timedelta)partial)WeakValueDictionary) ExceptionInfo)dumpsloadsprepare_accept_content)registry) bytes_to_str ensure_bytes)maybe_sanitize_url) current_appgroupmaybe_signaturestates)get_current_task)Context)BackendGetMetaErrorBackendStoreError ChordErrorImproperlyConfigured NotRegistered SecurityErrorTaskRevokedError TimeoutError) GroupResult ResultBase ResultSetallow_join_resultresult_from_tuple) BufferMap)LRUCache arity_greater) get_logger)create_exception_clsensure_serializableget_pickleable_exceptionget_pickled_exceptionraise_with_context) get_exponential_backoff_interval) BaseBackendKeyValueStoreBackendDisabledBackendpicklei pending_results_t)ZconcreteweakzU No result backend is configured. Please see the documentation for more information. z Starting chords requires a result backend to be configured. Note that a group chained with a task is also upgraded to be a chord, as this pattern requires synchronization. Result backends that supports chords: Redis, Database, Memcached, and more. cCs||dti|S)zReturn an unpickled backend.app)rZ_get_current_object)clsargskwargsr68/tmp/pip-unpacked-wheel-mu1yl971/celery/backends/base.pyunpickle_backendEsr8c@s eZdZddZeZZZdS) _nulldictcOsdSNr6)selfakwr6r6r7ignoreKsz_nulldict.ignoreN)__name__ __module__ __qualname__r> __setitem__update setdefaultr6r6r6r7r9Jsr9cCs|dkr dS|jSNF)Z ignore_resultrequestr6r6r7_is_request_ignore_resultQsrHc@s*eZdZejZejZejZeZdZdZ dZ dZ dddddZ dpdd Z dqd d Zd d ZddejfddZddddejfddZddZdddejfddZdddejfddZdrddZdsddZdtddZdd Zd!d"Zd#d$Zd%d&Zd'd(Z d)d*Z!d+d,Z"dud-d.Z#dvd/d0Z$d1d2Z%d3d4Z&dwd5d6Z'd7d8Z(dxd9d:Z)d;d<Z*d=d>Z+d?d@Z,e,Z-dAdBZ.dCdDZ/dEdFZ0dGdHZ1dIdJZ2dydKdLZ3dMdNZ4dOdPZ5dzdQdRZ6d{dSdTZ7dUdVZ8dWdXZ9dYdZZ:d[d\Z;d]d^ZdcddZ?d|dedfZ@dgdhZAdidjZBd}dkdlZCd~dndoZDdS)BackendNFTr) max_retriesZinterval_startZ interval_stepZ interval_maxc Ks||_|jj} |p| j|_tj|j\|_|_|_|p:| j } | dkrJt nt | d|_ | |||_|dkrr| jn||_|jdkr| jn|j|_t|j|_| dd|_| dd|_| dd|_| d td |_tit|_tt|_||_dS) N)limitZresult_backend_always_retryFZ+result_backend_max_sleep_between_retries_msi'Z,result_backend_base_sleep_between_retries_ms Zresult_backend_max_retriesinf) r2confZresult_serializer serializerserializer_registryZ _encoders content_typecontent_encodingencoderZresult_cache_maxr9r#_cacheprepare_expiresexpiresZresult_accept_contentacceptZaccept_contentr get always_retrymax_sleep_between_retries_msbase_sleep_between_retries_msfloatrLr0rZ_pending_resultsr"MESSAGE_BUFFER_MAXZ_pending_messagesurl) r;r2rRZmax_cached_resultsrZrYZ expires_typerar5rQZcmaxr6r6r7__init__us(     zBackend.__init__cCs2|r |jSt|jpd}|dr.|ddS|S)z=Return the backend as an URI, sanitizing the password or not.z:///NrM)rarendswith)r;Zinclude_passwordrar6r6r7as_uriszBackend.as_uricKs|||tjS)zMark a task as started.) store_resultrZSTARTEDr;task_idmetar6r6r7mark_as_startedszBackend.mark_as_startedcCs:|rt|s|j||||d|r6|jr6||||dS)z#Mark task as successfully executed.rFN)rHrfchordon_chord_part_return)r;rhresultrGrfstater6r6r7 mark_as_dones  zBackend.mark_as_donec Cs|r|j|||||d|r|jr0||||zt|j}Wnttfk r\t}YnX|D]z} t| } | | j | j d| _ | j d| _ |r|tjkr| jdk r|j| j|||| dd| j krb|| ||qb|r|jr||||dS)z#Mark task as executed with failure. tracebackrGrhgroup_idNrk)rfrkrliterchainAttributeError TypeErrortuplerrCoptionsr[idrrZPROPAGATE_STATESrherrbacks_call_task_errbacks) r;rhexcrqrGrfZ call_errbacksrnZ chain_dataZ chain_elemZchain_elem_ctxr6r6r7mark_as_failuresD      zBackend.mark_as_failurec Csg}|jD]}|j|}|js(|j|_zDt|jdr`t|jjts`t |jjdr`||||n | |Wq t k r| |Yq Xq |r|j }|j p|}t||jd}|jjjs|jddr|j|f||dn|j|f||ddS)N __header__rKr2Zis_eagerF) parent_idroot_id)rzr2 signatureZ_apphasattrtype isinstancer~rr$appendrryrrrQtask_always_eager delivery_infor[apply apply_async) r;rGr|rqZ old_signatureZerrbackrhrgr6r6r7r{s>       zBackend._call_task_errbacksrccCs<t|}|r |j|||d|d|r8|jr8||||dS)Nrp)rrfrkrl)r;rhreasonrGrfrnr|r6r6r7mark_as_revoked s  zBackend.mark_as_revokedcCs|j|||||dS)zfMark task as being retries. Note: Stores the current exception (if any). rp)rf)r;rhr|rqrGrfrnr6r6r7 mark_as_retrys zBackend.mark_as_retryc Cs|j}z|j|jj}Wntk r0|}YnXt|jd|jdgtd|}z| ||dWn6t k r}z|j |j |dWYSd}~XYnX|j |j |dSdS)NrhZ link_error)ryrzr)r|) r2Z_taskstaskbackendKeyErrorrrxr[dictr{ Exceptionfail_from_current_stackry)r;callbackr|r2rZ fake_requestZeb_excr6r6r7chord_error_from_stacks"   &zBackend.chord_error_from_stackcCst\}}}z6|dkr|n|}t|||f}||||j |WS|dk rz|j|jjWntk rxYnX|j}qF~XdSr:) sysexc_infotb_frameclearf_locals RuntimeErrortb_nextrr}rq)r;rhr|type_Zreal_exctbZexception_infor6r6r7r7s  zBackend.fail_from_current_stackcCsL|dkr|jn|}|tkr"t|St|}t|d|jt|j|j|j dS)z$Prepare exception for serialization.NrA)exc_type exc_message exc_module) rREXCEPTION_ABLE_CODECSr(rgetattrr?r'r4encoder@)r;r|rRexctyper6r6r7prepare_exceptionJs  zBackend.prepare_exceptionc Cs|sdSt|tr(|jtkr$t|}|St|tsrz t|}Wn2tk rp}ztd||W5d}~XYnX|d}z |d}Wn,tk r}zt d|W5d}~XYnX|dkrt |t }nRz(t j |}|dD]}t||}qWn(ttfk rt |tjj }YnX|dd}t|tr@t|tsp|dkrN|n |d|}td |d |z&t|ttfr||}n||}Wn8tk r} zt|d |d }W5d} ~ XYnX|S) z1Convert serialized exception to Python exception.NzbIf the stored exception isn't an instance of BaseException, it must be a dictionary. Instead got: rrz4Exception information must includethe exception type.rrcz!Expected an exception class, got z with payload ())r BaseExceptionrRrr)rrvr[r ValueErrorr&r?rmodulessplitrrucelery exceptionsr issubclassrrwlistr) r;r|errr3nameexc_msgZ fake_exc_typeerrr6r6r7exception_to_pythonTs\            &zBackend.exception_to_pythoncCs |jdkrt|tr|S|S)zPrepare value for storage.r/)rRrras_tupler;rmr6r6r7 prepare_valueszBackend.prepare_valuecCs||\}}}|Sr:)_encode)r;data_payloadr6r6r7rszBackend.encodecCst||jdS)N)rR)rrR)r;rr6r6r7rszBackend._encodecCs$|d|jkr ||d|d<|S)Nstatusrm)EXCEPTION_STATESr)r;rir6r6r7meta_from_decodedszBackend.meta_from_decodedcCs|||Sr:)rdecoder;rr6r6r7 decode_resultszBackend.decode_resultcCs.|dkr |S|pt|}t||j|j|jdS)N)rTrUrZ)strr rTrUrZrr6r6r7rs zBackend.decodecCs<|dkr|jjj}t|tr$|}|dk r8|r8||S|Sr:)r2rQZresult_expiresrr total_seconds)r;valuerr6r6r7rXs   zBackend.prepare_expirescCs(|dk r |S|jjj}|dkr$|jS|Sr:)r2rQZresult_persistent persistent)r;Zenabledrr6r6r7prepare_persistents zBackend.prepare_persistentcCs(||jkrt|tr||S||Sr:)rrrrr)r;rmrnr6r6r7 encode_results zBackend.encode_resultcCs ||jkSr:)rWr;rhr6r6r7 is_cachedszBackend.is_cachedcCs ||jkr t}|r$|}nd}||||||d}|rTt|ddrT|j|d<|rnt|ddrn|j|d<|jj ddr|rt|ddt|ddt|d dt|d dt|d dt |d r|j r|j d ndd} |rdd h} | D]"} | | } || } t| | | <q|| |S)N)rrmrqchildren date_donerrrrextendedrmrr4r5hostnameretriesrZ routing_key)rr4r5Zworkerrqueue) READY_STATESrutcnow isoformatcurrent_task_childrenrrrr2rQZfind_value_for_keyrrr[rr rC)r;rmrnrqrGZ format_daterrriZ request_metaZencode_needed_fieldsfieldrZ encoded_valuer6r6r7_get_result_metasH            zBackend._get_result_metacCst|dSr:)timesleep)r;amountr6r6r7_sleepszBackend._sleepc Ks|||}d}z"|j||||fd|i||WStk r}z^|jr||r||jkr|d7}t|j||jdd} | | qt t d||dnW5d}~XYqXqdS) zUpdate task state and result. if always_retry_backend_operation is activated, in the event of a recoverable exception, then retry operation with an exponential backoff until a limit has been reached. rrGrKTz%failed to store result on the backend)rhrnN) r _store_resultrr\exception_safe_to_retryrLr+r^r]rr*r) r;rhrmrnrqrGr5rr| sleep_amountr6r6r7rfs4    zBackend.store_resultcCs|j|d||dSr:)rWpop_forgetrr6r6r7forget%szBackend.forgetcCs tddS)Nz"backend does not implement forget.NotImplementedErrorrr6r6r7r)szBackend._forgetcCs||dS)zGet the state of a task.r) get_task_metarr6r6r7 get_state,szBackend.get_statecCs||dS)z$Get the traceback for a failed task.rqrr[rr6r6r7 get_traceback2szBackend.get_tracebackcCs||dS)zGet the result of a task.rmrrr6r6r7 get_result6szBackend.get_resultcCs*z||dWStk r$YnXdS)z(Get the list of subtasks sent by a task.rN)rrrr6r6r7 get_children:szBackend.get_childrencCs|jjjrtdtdS)Nz9Shouldn't retrieve result with task_always_eager enabled.)r2rQrwarningswarnRuntimeWarningr;r6r6r7_ensure_not_eagerAs  zBackend._ensure_not_eagercCsdS)aCheck if an exception is safe to retry. Backends have to overload this method with correct predicates dealing with their exceptions. By default no exception is safe to retry, it's up to backend implementation to define which exceptions are safe. Fr6)r;r|r6r6r7rHszBackend.exception_safe_to_retryc Cs||r.z |j|WStk r,YnXd}z||}WqWq2tk r}z\|jr||r||jkr|d7}t|j ||j dd}| |qt t d|dnW5d}~XYq2Xq2|r|dtjkr||j|<|S) zGet task meta from backend. if always_retry_backend_operation is activated, in the event of a recoverable exception, then retry operation with an exponential backoff until a limit has been reached. rrKTrzfailed to get meta)rhNr)rrWr_get_task_meta_forrr\rrLr+r^r]rr*rr[rSUCCESS)r;rhcacherrir|rr6r6r7rRs<      zBackend.get_task_metacCs|j|dd|j|<dS)z;Reload task result, even if it has been previously fetched.FrN)rrWrr6r6r7reload_task_resultyszBackend.reload_task_resultcCs|j|dd|j|<dS)zsz1Backend.fallback_chord_unlock..rmrrpriorityrzcelery.chord_unlock) countdownrr ) rrrxr[r2ZamqpZrouterZrouterZtasksrry)r; header_resultbodyr r5Z body_typerr r6r6r7fallback_chord_unlocks   zBackend.fallback_chord_unlockcCsdSr:r6rr6r6r7ensure_chords_allowedszBackend.ensure_chords_allowedcKs(||jj|}|j||f|dSr:)rr2rrr;Zheader_result_argsrr5rr6r6r7 apply_chords zBackend.apply_chordcCs0|pttdd}|r,ddt|dgDSdS)NrGcSsg|] }|qSr6rrr6r6r7r sz1Backend.current_task_children..r)rr)r;rGr6r6r7rszBackend.current_task_childrenr6cCs|sin|}t|j||ffSr:)r8 __class__r;r4r5r6r6r7 __reduce__s zBackend.__reduce__)NNNNNN)F)N)N)N)N)N)TF)NN)T)T)T)rK)N)r6N)Er?r@rArrZUNREADY_STATESrrZsubpolling_intervalsupports_native_joinZsupports_autoexpirerZ retry_policyrbrerjrroFAILUREr}r{ZREVOKEDrRETRYrrrrrrrrrrrrXrrrrrrfrrr get_statusrrrrrrrrrrrrrrrrrlrrrrrrr6r6r6r7rIWs    8.    G  0 " '   rIc@sHeZdZdddZdddZddd Zdd d Zd dZeddZ dS)SyncBackendMixinN?Tc csl||j}|sdSt}|D]*}t|tr>|j|jfVq ||jq |j||||||dEdHdS)N)timeoutintervalno_ack on_message on_interval)rresultssetrrryaddget_many) r;rmrrrr r!r"task_idsr6r6r7 iter_natives" zSyncBackendMixin.iter_nativec CsN||dk rtd|j|j||||d} | rJ|| |j||dSdS)Nz,Backend does not support on_message callback)rrr!r) propagater)rrwait_forryZ_maybe_set_cacheZ maybe_throw) r;rmrrrr r!rr(rir6r6r7wait_for_pendings z!SyncBackendMixin.wait_for_pendingcCs^|d}||}|dtjkr(|S|r2|t|||7}|r ||kr tdq dS)aLWait for task and return its result. If the task raises an exception, this exception will be re-raised by :func:`wait_for`. Raises: celery.exceptions.TimeoutError: If `timeout` is not :const:`None`, and the operation takes longer than `timeout` seconds. grzThe operation timed out.N)rrrrrrr)r;rhrrrr!Z time_elapsedrir6r6r7r)s    zSyncBackendMixin.wait_forFcCs|Sr:r6)r;rmr1r6r6r7add_pending_resultsz#SyncBackendMixin.add_pending_resultcCs|Sr:r6rr6r6r7remove_pending_resultsz&SyncBackendMixin.remove_pending_resultcCsdSrEr6rr6r6r7is_async szSyncBackendMixin.is_async)NrTNN)NrTNNNT)NrTN)F) r?r@rAr'r*r)r+r,propertyr-r6r6r6r7rs(    rc@seZdZdZdS)r,z"Base (synchronous) result backend.Nr?r@rA__doc__r6r6r6r7r,%sr,cseZdZeZdZdZdZdZfddZ ddZ d d Z d d Z d dZ ddZddZddZddZd9ddZd:ddZd;ddZddZejfd d!Zejfd"d#Zd$d%d&d$d$d$ejfd'd(Zd)d*Zdt|jdr|jj|_|tj|||jr:|j|_dS)N__func__) rkey_tr2_encode_prefixessuperrbimplements_incr_apply_chord_incrrrrr6r7rb3s   z!BaseKeyValueStoreBackend.__init__cCs.||j|_||j|_||j|_dSr:)r3task_keyprefixgroup_keyprefixchord_keyprefixrr6r6r7r4;sz)BaseKeyValueStoreBackend._encode_prefixescCs tddS)NzMust implement the get method.rr;keyr6r6r7r[@szBaseKeyValueStoreBackend.getcCs tddS)NzDoes not support get_manyr)r;keysr6r6r7mgetCszBaseKeyValueStoreBackend.mgetcCs |||Sr:)r#)r;r=rrnr6r6r7_set_with_stateFsz(BaseKeyValueStoreBackend._set_with_statecCs tddS)NzMust implement the set method.rr;r=rr6r6r7r#IszBaseKeyValueStoreBackend.setcCs tddS)Nz Must implement the delete methodrr<r6r6r7deleteLszBaseKeyValueStoreBackend.deletecCs tddS)NzDoes not implement incrrr<r6r6r7incrOszBaseKeyValueStoreBackend.incrcCsdSr:r6rAr6r6r7expireRszBaseKeyValueStoreBackend.expirerccCs$|j}|d|j||||gS)z#Get the cache key for a task by id.rc)r3joinr9)r;rhr=r3r6r6r7get_key_for_taskUs z)BaseKeyValueStoreBackend.get_key_for_taskcCs$|j}|d|j||||gS)z$Get the cache key for a group by id.rc)r3rEr:r;rrr=r3r6r6r7get_key_for_group\s z*BaseKeyValueStoreBackend.get_key_for_groupcCs$|j}|d|j||||gS)z?Get the cache key for the chord waiting on group with given id.rc)r3rEr;rGr6r6r7get_key_for_chordcs z*BaseKeyValueStoreBackend.get_key_for_chordcCsF||}|j|jfD]&}||rt|t|dSqt|S)zTake bytes: emit string.N)r3r9r: startswithr len)r;r=prefixr6r6r7 _strip_prefixjs   z&BaseKeyValueStoreBackend._strip_prefixccs:|D]0\}}|dk r||}|d|kr||fVqdS)Nr)r)r;valuesrkrr6r6r7 _filter_readyrs    z&BaseKeyValueStoreBackend._filter_readycsJt|dr(fdd||DSfddt||DSdS)Nitemscsi|]\}}||qSr6)rM)r rOvrr6r7 |sz=BaseKeyValueStoreBackend._mget_to_results..csi|]\}}t||qSr6r )r irR)r>r6r7rSs )rrPrQ enumerate)r;rNr>rr6)r>r;r7_mget_to_resultsys   z)BaseKeyValueStoreBackend._mget_to_resultsNrTc  #sf|dkr dn|}t|tr|nt|} t} j} | D]J} z | | } Wntk rZYq6X| d|kr6t| | fV| | q6| | d}| rbt| } fdd|D||}| || dd|D| D]&\}}|dk r||t||fVq|r2|||kr2t d|d |r>|t ||d 7}|r||krqbqdS) Nrrrcsg|]}|qSr6)rF)r rOrr6r7r sz5BaseKeyValueStoreBackend.get_many..cSsh|] }t|qSr6rT)r rRr6r6r7 sz4BaseKeyValueStoreBackend.get_many..zOperation timed out (rrK)rr#rWrr r$difference_updaterrWr?rCrQrrr)r;r&rrrr r!Zmax_iterationsridsZ cached_idsrrhcachedZ iterationsr>r r=rr6rr7r%sH       z!BaseKeyValueStoreBackend.get_manycCs|||dSr:)rBrFrr6r6r7rsz BaseKeyValueStoreBackend._forgetc Ks|j||||d}t||d<||}|dtjkr:|Sz||||||Wn6tk r} ztt | ||d| W5d} ~ XYnX|S)N)rmrnrqrGrhr)rnrh) rr rrrr@rFrrr) r;rhrmrnrqrGr5riZ current_metaexr6r6r7rs  &z&BaseKeyValueStoreBackend._store_resultcCs(||||d|itj|S)Nrm)r@rHrrrrrr6r6r7rs  z$BaseKeyValueStoreBackend._save_groupcCs|||dSr:)rBrHrr6r6r7rsz&BaseKeyValueStoreBackend._delete_groupcCs*|||}|s tjddS||S)$Get task meta-data for a task by id.N)rrm)r[rFrZPENDINGrrgr6r6r7rs z+BaseKeyValueStoreBackend._get_task_meta_forcCs>|||}|r:||}|d}t||j|d<|SdS)r]rmN)r[rHrr!r2)r;rrrirmr6r6r7rs  z'BaseKeyValueStoreBackend._restore_groupcKs$||jj|}|j|ddS)Nr)rr2rsaverr6r6r7r7s z*BaseKeyValueStoreBackend._apply_chord_incrc Ks|js dS|j}|j}|sdS||}ztj||d}WnXtk r} z:t|j|d} t d|| | | t d| WYSd} ~ XYnX|dkrz t |Wn\t k r} zszDisabledBackend._is_disabledcOsdS)Nz disabled://r6rr6r6r7reAszDisabledBackend.as_uriN)r?r@rAr0rWrfrrtrerrrrZget_task_meta_forr)r%r6r6r6r7r.3sr.)Tr0rrr collectionsrrr functoolsrweakrefrZbilliard.einforZkombu.serializationrr r r rSZkombu.utils.encodingr r Zkombu.utils.urlrZcelery.exceptionsrrrrrZ celery._staterZcelery.app.taskrrrrrrrrrZ celery.resultrrrr r!Zcelery.utils.collectionsr"Zcelery.utils.functionalr#r$Zcelery.utils.logr%Zcelery.utils.serializationr&r'r(r)r*Zcelery.utils.timer+__all__ frozensetrr?rar`r0rsrqr8rr9rHrIrr,ZBaseDictBackendr1r-r.r6r6r6r7s^        (      N