`f@v dZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z ddl mZddl mZdd lmZdd lmZdd lmZmZdd lmZdd lmZddlmZddlmZddlm Z ddl!m"Z"m#Z#ddl$m%Z%ddl&m'Z'ddl(m)Z)ddl(m*Z+ddl(m,Z,m-Z-m.Z.m/Z/m0Z0ddl1m2Z2ddl3m4Z4m5Z5dZ6dZ7dZ8dZ9d;dZ:Gdd e!Z;e%j<j=Gd"d#e>Z?d$Z@e?Ad%&Gd'd(e?ZBGd)d%eBZCGd*d+e?ZDe?AGd,d-eDZEe?AGd.d/eDZFe?AGd0d1e?Z*d2ZGe?AGd3d4e?ZHe?Ad5&Gd6d7e?ZIeIZJd8ZKeKZLd>> d1 = {'dict': {'a': 1}, 'list': [1, 2], 'tuple': (1, 2)} >>> d2 = {'dict': {'b': 2}, 'list': [3, 4], 'set': {'a', 'b'}} >>> _merge_dictionaries(d1, d2) d1 will be modified to: { 'dict': {'a': 1, 'b': 2}, 'list': [1, 2, 3, 4], 'tuple': (1, 2), 'set': {'a', 'b'} } Arguments: d1 (dict): Dictionary to merge into. d2 (dict): Dictionary to merge from. aggregate_duplicates (bool): If True, aggregate duplicated items (by key) into a list of all values in d1 in the same key. If False, duplicate keys will be taken from d2 and override the value in d1. N) items isinstancedict_merge_dictionariesintfloatstrr0extendappend)d1d2aggregate_duplicateskeyvalues r3rFrFIsZ, hhjj,, U "99%&& ,#BsGRW5555ec5#%677I)=Hugg5BsGbgt,,,BsGT1J1J,sGNN2c7++++),#w"$3"&r#w--3sGNN2c7+++hhjj U b==BsGr;ceZdZdZdefdZddZdefdZddZe defdZ defd Z dd Z defd Z defd Zdefd ZdS)StampingVisitorzStamping API. A class that provides a stamping API possibility for canvas primitives. If you want to implement stamping behavior for a canvas primitive override method that represents it. returnc iS)aMethod that is called on group stamping start. Arguments: group (group): Group that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. selfr&r@s r3on_group_startzStampingVisitor.on_group_start|  r;Nc dS)zMethod that is called on group stamping end. Arguments: group (group): Group that is stamped. headers (Dict): Partial headers that could be merged with existing headers. NrUrVs r3 on_group_endzStampingVisitor.on_group_end  r;c iS)aMethod that is called on chain stamping start. Arguments: chain (chain): Chain that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. rUrWr#r@s r3on_chain_startzStampingVisitor.on_chain_startrYr;c dS)zMethod that is called on chain stamping end. Arguments: chain (chain): Chain that is stamped. headers (Dict): Partial headers that could be merged with existing headers. NrUr^s r3 on_chain_endzStampingVisitor.on_chain_endr\r;c dS)aMethod that is called on signature stamping. Arguments: sig (Signature): Signature that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. NrU)rWsigr@s r3 on_signaturezStampingVisitor.on_signaturesr;c t|jtst|j|_|j|jfi|S)uMethod that is called on сhord header stamping start. Arguments: sig (chord): chord that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. )rDr-r&rXrWrcheaders r3on_chord_header_startz%StampingVisitor.on_chord_header_startsF#)U++ )ci((CI"t"3977777r;c ,|j|jfi|dS)uMethod that is called on сhord header stamping end. Arguments: sig (chord): chord that is stamped. headers (Dict): Partial headers that could be merged with existing headers. N)r[r-rfs r3on_chord_header_endz#StampingVisitor.on_chord_header_ends' #)..v.....r;c iS)a Method that is called on chord body stamping. Arguments: sig (chord): chord that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. rUrfs r3 on_chord_bodyzStampingVisitor.on_chord_bodyrYr;c iS)aMethod that is called on callback stamping. Arguments: callback (Signature): callback that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. rU)rWcallbackrgs r3 on_callbackzStampingVisitor.on_callbackrYr;c iS)aMethod that is called on errback stamping. Arguments: errback (Signature): errback that is stamped. headers (Dict): Partial headers that could be merged with existing headers. Returns: Dict: headers to update. rU)rWerrbackrgs r3 on_errbackzStampingVisitor.on_errbackrYr;)rSN)__name__ __module__ __qualname____doc__rErXr[r_rarrdrhrjrlrorrrUr;r3rRrRvs# $         $         d   ^  8d 8 8 8 8//// d          t      r;rR) metaclassceZdZdZiZdxZZddhZed?dZ ed?dZ d@fd Z d Z d Z dAd ZdBd ZdCd ZdAdZeZ dDdZeZdBdZd?dZdZdEdZdFdZdGdZdZdZdZdZdZdZ dZ!dZ"dZ#d Z$d!Z%d"Z&d#Z'd$Z(d%Z)d&Z*fd'Z+e,d(Z-e.d)Z/e.d*Z0e.d+Z1e.d,Z2e3d-d.Z4e3d/d0Z5e3d1d2Z6e3d3d4Z7e3d5d6Z8e3d7d8Z9e3d9d:Z:e3d;d<Z;e3d=d>Z<xZ=S)Hr"a/Task Signature. Class that wraps the arguments and execution options for a single task invocation. Used as the parts in a :class:`group` and other constructs, or to pass tasks around as callbacks while being compatible with serializers with a strict type subset. Signatures can also be created from tasks: - Using the ``.signature()`` method that has the same signature as ``Task.apply_async``: .. code-block:: pycon >>> add.signature(args=(1,), kwargs={'kw': 2}, options={}) - or the ``.s()`` shortcut that works for star arguments: .. code-block:: pycon >>> add.s(1, kw=2) - the ``.s()`` shortcut does not allow you to specify execution options but there's a chaining `.set` method that returns the signature: .. code-block:: pycon >>> add.s(2, 2).set(countdown=10).set(expires=30).delay() Note: You should use :func:`~celery.signature` to create new signatures. The ``Signature`` class is the type returned by that function and should be used for ``isinstance`` checks for signatures. See Also: :ref:`guide-canvas` for the complete guide. Arguments: task (Union[Type[celery.app.task.Task], str]): Either a task class/instance, or the name of a task. args (Tuple): Positional arguments to apply. kwargs (Dict): Keyword arguments to apply. options (Dict): Additional options to :meth:`Task.apply_async`. Note: If the first argument is a :class:`dict`, the other arguments will be ignored and the values in the dict will be used instead:: >>> s = signature('tasks.add', args=(2, 2)) >>> signature(s) {'task': 'tasks.add', args=(2, 2), kwargs={}, options={}} Ngroup_idstamped_headerscfd}|S)zRegister a new type of signature. Used as a class decorator, for example: >>> @Signature.register_type() >>> class mysig(Signature): >>> pass c*|jp|j<|SN)TYPESrs)subclassclsr6s r3_innerz'Signature.register_type.._inner/s3;CId/h/ 0Or;rU)rr6rs`` r3 register_typezSignature.register_type's)       r;c|d}|r(|j|}||ur|||St||S)zCreate a new signature from a dict. Subclasses can override this method to customize how are they created from a dict. subtask_typeapp)getr~ from_dictr")rdrtyp target_clss r3rzSignature.from_dict5s_ eeN##  83J$$!++A3+777$$$$r;Fc T||_t|tr#t|dS |j} ||_n#t$r|} YnwxYwt| t|pd|pit|pifi| ||dS)NrUr9argskwargsoptionsr immutable) _apprDrEsuper__init__r6_typer1tuple) rWr9rrrtyperrrex task_name __class__s r3rzSignature.__init__Bs dD ! !  GG  T " " " " " " I " " ! ! !  ! GG  U4:2%6%6|W]11b11)#      sA A A cR|||d\}}}|j|i|S)z0Call the task directly (in the current process).N)_merger)rW partial_argspartial_kwargsrr_s r3__call__zSignature.__call__Zs4++lNDIIfaty$)&)))r;c.|||S)z5Shortcut to :meth:`apply_async` using star arguments. apply_async)rWrrs r3delayzSignature.delay_s n===r;c |r|nd}|r|ni}d|D}||||\}}}|jj||fi|S)zCall task locally. Same as :meth:`apply_async` but executed the task inline instead of sending a task message. rUci|] \}}||| Sr}rU.0kvs r3 z#Signature.apply..lEEEDAqq}1a}}}r;)rCrrapply)rWrrrs r3rzSignature.applycsu #tt!)rEEGMMOOEEE $ D&' B BfgtytV77w777r;c |r|nd}|r|ni}d|D} |j}n#t$rYdSwxYw|s|s|r||||\}}}n|j|j|j}}}|||fi|S)aApply this task asynchronously. Arguments: args (Tuple): Partial args to be prepended to the existing args. kwargs (Dict): Partial kwargs to be merged with existing kwargs. options (Dict): Partial options to be merged with existing options. Returns: ~@AsyncResult: promise of future evaluation. See also: :meth:`~@Task.apply_async` and the :ref:`guide-calling` guide. rUci|] \}}||| Sr}rUrs r3rz)Signature.apply_async..rr;N)rC _apply_async IndexErrorrrrr)rWrr route_namer_applys r3rzSignature.apply_asyncqs#tt!)rEEGMMOOEEE &FF    FF   I6 IW I$(KKfg$F$F !D&''$(It{DL'&DvdF..g...s4 AAc|r|nd}|r|ni}|{jdjvr@jtjdgijfd|D}nj}jr|sjj|fS|r$t|tjznj|rtjfi|nj|fS)aGMerge partial args/kwargs/options with existing ones. If the signature is immutable and ``force`` is False, the existing args/kwargs will be returned as-is and only the options will be merged. Stamped headers are considered immutable and will not be merged regardless. Arguments: args (Tuple): Partial args to be prepended to the existing args. kwargs (Dict): Partial kwargs to be merged with existing kwargs. options (Dict): Partial options to be merged with existing options. force (bool): If True, the args/kwargs will be merged even if the signature is immutable. The stamped headers are not affected by this option and will not be merged regardless. Returns: Tuple: (args, kwargs, options) rUNrzc6i|]\}}|vs |jv||SrU)r)rrrimmutable_optionsrWs r3rz$Signature._merge..sA...A---$,1F1F11F1F1Fr;) _IMMUTABLE_OPTIONSrunionsetrrCrrrrrE)rWrrrforce new_optionsrs` @r3rzSignature._merges:&#tt!)r   !% 7  DL00$($;$A$A#dlFVFVWhjlFmFmBnBn$o$o!T\.....!(...KK ,K > 9% 9It{K8 826Ed eDI....DI/5FT[++F+++4; r;c F|r|nd}|r|ni}|s|s|r||||\}}}n|j|j|j}}}t|jt||t||j |j d|j }|j |_ |S)a7Create a copy of this signature. Arguments: args (Tuple): Partial args to be prepended to the existing args. kwargs (Dict): Partial kwargs to be merged with existing kwargs. options (Dict): Partial options to be merged with existing options. rUrr) rrrrr"rr9rrrrrr)rWrroptsr(s r3clonezSignature.clones#tt!)r  F6 FT F!%T64!@!@ D&$$!%DK$&D''16t394.]s/iiiAgVgNhIhIh1IhIhIhr;) copyr0keysrFrrrrrC)rWvisitor_headersr? self_headersr@rzredacted_optionss ` r3_stamp_headerszSignature._stamp_headers)st>,,.. G + +)-gllnn)=)=G% &  &-3O 7759/:N:N:P:P5Q5Q 12 } ] ] ] ])-c':K2L.M.M)N)NG% &  O!'++.?"D"DEEO  " "4<#3#34Er#J#J K K K)-o)>)>G% &iiii1C1C1E1Eiii  )9P] ^ ^ ^ ^)-c':K2L.M.M)N)NG% &r;c |j||fi||}d}||j|fi|pi}|j||fi|}|jdi|S)aStamp this signature with additional custom headers. Using a visitor will pass on responsibility for the stamping to the visitor. .. versionadded:: 5.3 Arguments: visitor (StampingVisitor): Visitor API object. append_stamps (bool): If True, duplicated stamps will be appended to a list. If False, duplicated stamps will be replaced by the last stamp. headers (Dict): Stamps that should be added to headers. NrU) stamp_linksrrdrr)rWr>r?r@rs r3r=zSignature.stampes -;;7;;;,,..  2g24CC7CCIrO%$%o}PPPPtx""'"""r;c F|}d}t|}t|jdpgD]L}t ||j}d}||j|fi|pi}|jd|||d|}|j ||fi|Mt|}t|jdpgD]L}t ||j}d}||j |fi|pi}|jd|||d|}|j ||fi|MdS)aStamp this signature links (callbacks and errbacks). Using a visitor will pass on responsibility for the stamping to the visitor. Arguments: visitor (StampingVisitor): Visitor API object. append_stamps (bool): If True, duplicated stamps will be appended to a list. If False, duplicated stamps will be replaced by the last stamp. headers (Dict): Stamps that should be added to headers. FlinkrN)rr?r link_errorrU) rrrrrr)rrorr=rr)rWr>r?r@non_visitor_headersrrrs r3rzSignature.stamp_links{s&llnn  .//t|//7788>B : :D"4TX666D"O""5'"5d"F"Fg"F"F"L")d) /+) G DJw 9 9 9 9 9 9.//t|// ==>>D" : :D"4TX666D"O""4'"4T"E"EW"E"E"K)d) /+) G DJw 9 9 9 9 9 9 : :r;c|j|g}t|ts |gx}|j|<|S)aGets the value at the given self.options[key] as a list. If the value is not a list, it will be converted to one and saved in self.options. If the key does not exist, an empty list will be set and returned instead. Arguments: key (str): The key to get the value for. Returns: List: The value at the given key as a list or an empty list if the key does not exist. )r setdefaultrDr)rWrOrCs r3_with_list_optionzSignature._with_list_optionsE ''R00%11 0). /EDL% r;cb||}||vr|||S)zEAppends the given value to the list at the given key in self.options.)rrKrWrOrPrCs r3append_to_list_optionzSignature.append_to_list_options7&&s++    LL    r;ct||}|t|dS)zExtends the list at the given key in self.options with the given value. If the value is not a list, it will be converted to one. N)rrJrrs r3extend_list_optionzSignature.extend_list_options7 &&s++ Z&&'''''r;c.|d|S)zAdd callback task to be applied if this task succeeds. Returns: Signature: the argument passed, for chaining or use with :func:`~functools.reduce`. rrrWrns r3rzSignature.links))&(;;;r;c.|d|S)zAdd callback task to be applied on error in task execution. Returns: Signature: the argument passed, for chaining or use with :func:`~functools.reduce`. rrrWrqs r3rzSignature.link_errors)),@@@r;c0|||S)a2Version of :meth:`link_error` that supports chaining. on_error chains the original signature, not the errback so:: >>> add.s(2, 2).on_error(errback.s()).delay() calls the ``add`` task, not the ``errback`` task, but the reverse is true for :meth:`link_error`. )rrs r3on_errorzSignature.on_errors     r;c ttjtj|ggdt |jdpgDS)zhReturn a recursive list of dependencies. "unchain" if you will, but with links intact. c3>K|]}|VdSr}) flatten_links)rrs r3 z*Signature.flatten_links..sJ E E   ! ! E E E E E Er;r)r0 itertoolsr# from_iterablerrrrWs r3rzSignature.flatten_linkss} IO11)/VH E E#DL$4$4V$<$<==C E E E3 3      r;cvt|tr7tt|f||jSt|t r&t |}t|||jSt|trt|||jStS)zChaining operator. Example: >>> add.s(2, 2) | add.s(4) | add.s(8) Returns: chain: Constructs a :class:`~celery.canvas.chain` of the given signatures. r) rD_chainr unchain_tasksrr&r4rr"NotImplementedrWothers r3__or__zSignature.__or__s eV $ $ 6.,,..0059Y@@@ @ u % % 6&u--E$48444 4 y ) ) 6$49555 5r;c,||Sr})rrs r3__ior__zSignature.__ior__ s{{5!!!r;c |j}|j}|jdp t }|d5}|j||}|j |d|j dd|i||j | |cdddS#1swxYwYdS)Nrr9) connectionrU) rrrrrproducer_or_acquirebackend on_task_callcontrolelectionrrr)rWrrrproducerpropss r3rzSignature.electionsyhly))3TVV  $ $T * * )hL--h<>> add.s(2, 2).reprcall() 'add(2, 2)' T)rr9)rr )rWrrrs r3r zSignature.reprcalls9++dFBd+CCfaV dF333r;cD||t|<t|Sr})idrE)rWmemos r3 __deepcopy__zSignature.__deepcopy__-sRXXDzzr;cN|Sr})rrrs r3 __invert__zSignature.__invert__1s !!%%'''r;c0tt|ffSr})r(rErs r3 __reduce__zSignature.__reduce__4s4::-''r;c t|Sr})rErs r3__json__zSignature.__json__9sDzzr;c*|Sr})r rs r3__repr__zSignature.__repr__<s}}r;c#KtD]4\}}t|tr|n||fV5dSr})rrCrDbytesdecode)rWrrrs r3rCzSignature.items?s_GGMMOO ? ?DAq *1e 4 4;!((***!Q> > > > > ? ?r;c|jSr}r8rs r3r6zSignature.nameCs yr;c@|jp|jj|dSNr9)rrr-rs r3rzSignature.typeHsz9TX^DL99r;c|jptSr})rrrs r3rz Signature.appLsy'K'r;cT |jjS#t$r|jjcYSwxYwr})rrrrrs r3rzSignature.AsyncResultPs> (9( ( ( ( (8' ' ' ' (s ''c| |jjS#t$r#t|jj|dcYSwxYwr)rrr_partialr send_taskrs r3rzSignature._apply_asyncWsL >9( ( > > >DH.V == = = = >s *;;zoptions.task_idz Task UUIDzoptions.parent_idzTask parent UUID.zoptions.root_idzTask root UUID.r9z Name of task.rzPositional arguments to task.rzKeyword arguments to task.rzTask execution options.rzType of signaturerz+Flag set if no longer accepts new argumentsr})NNNNNNFNNN)NNN)NNNFNNNNNN)NFTNF)F)>rsrtrurvr~rrr classmethodrrrrrrrrrrr_freezerrrrr=rrrrrrrrrrrr r rrrrrCpropertyr6rrrrrrr rrr9rrrrr __classcell__rs@r3r"r"s66p ED5%&78   [  % % %[ %CG9>0*** >>> 8 8 8 8////B((((T4G489=/%/%/%/%bG        ###::::x####,/:/:/:/:b"(((<<<AAA       .""" ) ) ) 4 4 4(((((( ?????X::_:((_(((_( >>_>  +[ 9 9B  !46IJJI02CDDG  FO 4 4D  F$C D DD  h(D E EFy*CDDG##N4GHHL  BDDIIIIIr;r"c|rtddi|Sd|vrtd|i|S|td|d|zi|SdS)Nr#r)rr-use_links r3_prepare_chain_from_optionsr+jswR !$111   %('222  ''"2U":;!! !  r;r#)r6c eZdZeddZeddZfdZdZdZ fdZ d Z dd Z dd Z dd ZddZddddddddejdf dZddZedZdZxZS)r kwargs.taskszTasks in chain.Nc|dd}|r=t|trt|x}|dd<fd|D}||fdi|dS)Nrr-c2g|]}t|Srr)rr9rs r3 z$_chain.from_dict..s&FFF_Ts333FFFr;rr)rDrr0)rrrr-s ` r3rz_chain.from_dicts( G$  G%'' ;/3E{{:( G,FFFFFFFEs522c2Qy\222r;ct|dkr*t|drt|dn|}tjddd|ifi||dd|_d|_d|_dS)Nr+rz celery.chainrUr-r*r#) r,rrrrpop _use_linkr_frozenrWr-rrs r3rz_chain.__init__s$'JJ!OOa8I8IOuQx gu-=      Z66# r;c@|jr|||SdSr})r-r)rWrrs r3rz_chain.__call__s, : 2##D&11 1 2 2r;ct|trt|}|}|s|St|dtr:|dj|z|d_t |||jSt |t|||j St|trQt |t|||j St|tr|j r`t|j dtr@|}t |j d||j |j d<|S|j r[t|j dtr;|}|j dj|z|j d_|St |t|||j StS)Nr)rDr&r4rr'bodyrrrrrrr"r-rr)rWrr-rcs r3rz _chain.__or__s2 eU # #& "&u--E&&((E  %)U++ 7!&r%!7b !tDzz%TX66664::ou#'9... . v & & "4::n""$$e&9&9&;&;==BF)MMM M y ) ) "z AjB?? Ajjll %IbM5di!9!9!9 "   A 4:b>5 A A Ajjll%(Yr]%7%%? " " "tDzz/&&((%#1#16:iAAAA" !r;cttj|i|}fd|jdD|jd<|S)Nc6g|]}|jdS)T)rr)r)rrcrW to_signatures r3r3z _chain.clone..s<% % %  L$)4 8 8 8% % % r;r-)r)rrr)rWrrr(r?rs` @r3rz _chain.clonesi& !EGGM42622 % % % % %  '0% % % !r;cBd|jD}t|jdpgD]}|d|t|jdpgD]}|D]}|||S)aReturn a list of tasks in the chain. The tasks list would be cloned from the chain's tasks. All of the chain callbacks would be added to the last task in the (cloned) chain. All of the tasks would be linked to the same error callback as the chain itself, to ensure that the correct error callback is called if any of the (cloned) tasks of the chain fail. c6g|]}|SrUrrts r3r3z(_chain.unchain_tasks..s ///q///r;rr;r)r-rrrrr)rWr-rcr9s r3rz_chain.unchain_taskss0/DJ///dl..v6677=2  C "INN3    dl..|<<==C % %C % %$$$$ % r;c |r|nd}|r|ng}|j}|jjr5t5|j||fi|cdddS#1swxYwY|j||fd|i|rt |jfi|n|jS)NrUr)rconftask_always_eagerrrrunrEr)rWrrrrs r3rz_chain.apply_asyncs#tt!)rh 8 % ;"$$ ; ;!tz$::':: ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ;txfII#I-4 FD ) ) ) ) )$,II IsA  AAc |r|nd}|r|ng}| p|j} |j}|| jjdkrd}|r+|js$t |t |jzn|j}||||j| | || ||||  \}}|r\|r|d d|| }t|||}|j di|}|s|S|dSdS)zExecutes the chain. Responsible for executing the chain in the correct order. In a case of a chain of a single task, the task is executed directly and the result is returned for that task specifically. rUNr+T)rrr) rr6rF task_protocolrrr prepare_stepsr-rr5r+r)rWrrryr'rrr publisherrrrrrrr*r-results_from_prepare first_taskresult_from_applys r3rHz _chain.runsF#tt!)roTX>   6! ; ;H> $>d eDI....48I '+&8&8 &$*gy*c Xu+'9' ' ##  / :a++FD999J1'5(KKG 6 6 A A A A   /((+A.. / /r;c||j|j|j||d|j|||d| x\}}|_|dS)NF)rrr)rKrrr-rr7) rWrryr'rrrrresultss r3rz _chain.freeze>sZ%)$6$6 It{DJD Hc8U%#%7% %  7T\ qzr;Fc d}||j|fi|pi}|j||fi|}|j|fi||jD]}|j||fi|||j|fi|dSdSr})r_rrr-r=ra)rWr>r?r@rr9s r3r=z _chain.stampIs  4g4TEEWEEKO%$%o}PPPP,,G,,,J : :D DJw 9 9 9 9 9 9   G  1 1 1 1 1 1 1  r;Tc\|p|j}|j}||jjdkrd}t |}|j}|j}d}d}gg}}d}|r[|}| | }}t|tj s | ||}t|trt|}| r.|r| ||}n@| }n+|r)t|t|jz|_t|tr||jt|trh|rf|| t#|||j||}n##t&$rt#||||}YnwxYw|r|||| | | }n|| }|dz }|r'|r|||r|js||_|r't/|D]}||||||||}}t|t"r2|j|}|jr|j}|j|}|[||fS) aPrepare the chain for execution. To execute a chain, we first need to unpack it correctly. During the unpacking, we might encounter other chains, groups, or chords which we need to unpack as well. For example: chain(signature1, chain(signature2, signature3)) --> Upgrades to chain(signature1, signature2, signature3) chain(group(signature1, signature2), signature3) --> Upgrades to chord([signature1, signature2], signature3) The responsibility of this method is to ensure that the chain is correctly unpacked, and then the correct callbacks are set up along the way. Arguments: args (Tuple): Partial args to be prepended to the existing args. kwargs (Dict): Partial kwargs to be merged with existing kwargs. tasks (List[Signature]): The tasks of the chain. root_id (str): The id of the root task. parent_id (str): The id of the parent task. link_error (Union[List[Signature], Signature]): The error callback. will be set for all tasks in the chain. app (Celery): The Celery app instance. last_task_id (str): The id of the last task in the chain. group_id (str): The id of the group that the chain is a part of. chord_body (Signature): The body of the chord, used to synchronize with the chain's last task and the chord's body when used together. clone (bool): Whether to clone the chain's tasks before modifying them. from_dict (Callable): A function that takes a dict and returns a Signature. Returns: Tuple[List[Signature], List[AsyncResult]]: The frozen tasks of the chain, and the async results Nr+Trr)r<rrr)r<rr)rryr'rr)rr6rFrJrr5rJrDrCallableSignaturer&r4rrrrr-r'rr1rrparentrrrKrensure_chords_allowed)rWrrr-rrrr last_task_idry chord_bodyrrrr*steps steps_pop steps_extend prev_taskprev_resrQir9 is_first_task is_last_taskresrqnodes r3rKz_chain.prepare_stepsVsLJoTX >   6! ; ;He I |  Rw  \ 9;;D/4)U?? 0 y3///$&& 0*$// ; (::dF33DD::<rr9z | c34K|]}t|VdSr})reprrCs r3rz"_chain.__repr__.. s(331tAww333333r;)r-rrsr r joinrs r3rz_chain.__repr__sxz CBtDzz*BBRXXBBBB B) JqM& ! JJ33 333 3 355 5r;r}r!) NNNNNNNNNNNNNr"r#)rsrtrurr-r$rrrrrrrrHrr=r"rKrr&rrr'r(s@r3rrs  ^-> ? ?E333[3222'"'"'"R& I I I I@D@D=A,/,/,/,/`599=     2 2 2 2#dt#'$4 I,?"& ZZZZx""X"5555555r;rc"eZdZdZfdZxZS)r#aChain tasks together. Each tasks follows one another, by being applied as a callback of the previous task. Note: If called with only one argument, then that argument must be an iterable of tasks to chain: this allows us to use generator expressions. Example: This is effectively :math:`((2 + 2) + 4)`: .. code-block:: pycon >>> res = chain(add.s(2, 2), add.s(4))() >>> res.get() 8 Calling a chain will return the result of the last task in the chain. You can get to the other tasks by following the ``result.parent``'s: .. code-block:: pycon >>> res.parent.get() 4 Using a generator expression: .. code-block:: pycon >>> lazy_chain = chain(add.s(i) for i in range(10)) >>> res = lazy_chain(3) Arguments: *tasks (Signature): List of task signatures to chain. If only one argument is passed and that argument is an iterable, then that'll be used as the list of signatures to chain instead. This means that you can use a generator expression. Returns: ~celery.chain: A lazy signature that can be called to apply the first task in the chain. When that task succeeds the next task in the chain is applied, and so on. c|s|rt|dkst|drt|dkr|dn|}ttj|t }|t krEt|t r0t||s tj ||j fi|S|Stj |g|Ri|S)Nr+r) r,rr operatoror_rr#rDr__new__r-)rr-r new_instancers r3rwz chain.__new__?s $% $5zzQ'%("3"3$'JJ!OOa &hlE688DD %<> BF   r;r}r!) rsrtrurr rr$rrrr'r(s@r3rzrzMsJ:fd++LLLL[L         r;rzceZdZdZdZdZdS)r$zMap operation for tasks. Note: Tasks executed sequentially in process, this is not a parallel operation like :class:`group`. z celery.mapc||j\}}d|jdtt |ddS)N[z (x) for x in d]rrr9r!rqrWr9r{s r3rz xmap.__repr__psF$$T[11bE49EE8DHHc+B+BEEEEr;NrsrtrurvrrrUr;r3r$r$es9JFFFFFr;r$ceZdZdZdZdZdS)r%z.Map operation for tasks, using star arguments.zcelery.starmapc||j\}}d|jdtt |ddS)Nrz(*x) for x in rrrrs r3rzxstarmap.__repr__{sF$$T[11bF49FFHT"XXs,C,CFFFFr;NrrUr;r3r%r%us388!JGGGGGr;r%ceZdZdZedddZed dZfdZdZ d d Z d Z ed d Z xZ S)rz)Partition of tasks into chunks of size n.r9r{nNcT|||dd|i|dSr}r~rs r3rzchunks.from_dictrr;c ftjdd|t||dfddi|dS)Nz celery.chunksrU)r9r{rrT)rrr)rWr9r{rrrs r3rzchunks.__init__sT""&eBiia@@  #' +2     r;c |jdi|S)NrUr)rWrs r3rzchunks.__call__st**'***r;c |r|nd}|r|ni}|j||fdt|jdi|S)NrUrr9)r&rr:rr)rWrrrs r3rzchunks.apply_asyncsm#tt!)r'tzz||' &  %dkoof&=&=>> BF   r;cj\}}tfdtt ||DjS)Nc3FK|]}t|jVdSrN)r%r)rpartrWr9s r3rzchunks.group..sH88tTty999888888r;r)rrr&_chunksiterr)rWr{rr9s` @r3r&z chunks.groupsv'' 44 b!88888")$r((A"6"6888$$$ $r;c0|||||S)NrrU)rr9r{rrs r3 apply_chunkszchunks.apply_chunkss#(ss4QC(((***r;r}r!)rsrtrurvr rr$rrrrr&rr'r(s@r3rrs33:fdC00LLLL[L +++    $$$+++[+++++r;rcVt|trt|}t|ttfr|j}n]t|t jr|g}n?t|trtfd|D}nfd|D}|S)Nrc3:K|]}t|VdSrr(rrDrs r3rz_maybe_group..s0??A)A3///??????r;c2g|]}t|Sr0rrs r3r3z _maybe_group..s&:::qYqc***:::r;) rDrEr(r&rr-rrUr r)r-rs `r3 _maybe_grouprs%*%S)))%%));  E85 6 6; e] + + ;?????????EE::::E:::E Lr;ceZdZdZeddZeddZfdZdZ dZ dd Z d d Z d!d Z dZd"dZdZdZejejeefdZ d#dZdZ d$dZ d$dZeZdZdZdZdZ dZ!e"dZ#xZ$S)%r&aqCreates a group of tasks to be executed in parallel. A group is lazy so you must call it to take action and evaluate the group. Note: If only one argument is passed, and that argument is an iterable then that'll be used as the list of tasks instead: this allows us to use ``group`` with generator expressions. Example: >>> lazy_group = group([add.s(2, 2), add.s(4, 4)]) >>> promise = lazy_group() # <-- evaluate: returns lazy result. >>> promise.get() # <-- will wait for the task to return [4, 8] Arguments: *tasks (List[Signature]): A list of signatures that this group will call. If there's only one argument, and that argument is an iterable, then that'll define the list of signatures instead. **options (Any): Execution options applied to all tasks in the group. Returns: ~celery.group: signature that when called will then call all of the tasks in the group (and return a :class:`GroupResult` instance that can be used to inspect the state of the group). r-zTasks in group.Nc|dd}t|fd|Dx|dd<}||fdi|dS)aCreate a group signature from a dictionary that represents a group. Example: >>> group_dict = { "task": "celery.group", "args": [], "kwargs": { "tasks": [ { "task": "add", "args": [ 1, 2 ], "kwargs": {}, "options": {}, "subtask_type": None, "immutable": False }, { "task": "add", "args": [ 3, 4 ], "kwargs": {}, "options": {}, "subtask_type": None, "immutable": False } ] }, "options": {}, "subtask_type": "group", "immutable": False } >>> group_sig = group.from_dict(group_dict) Iterates over the given tasks in the dictionary and convert them to signatures. Tasks needs to be defined in d['kwargs']['tasks'] as a sequence of tasks. The tasks themselves can be dictionaries or signatures (or both). rr-c3:K|]}t|VdSrr1r2s r3rz"group.from_dict.. sI@ @ /3ODc * * *@ @ @ @ @ @ r;rr)r)rrr orig_tasks rebuilt_taskss ` r3rzgroup.from_dictsdx[) /?tJ/?/?@ @ @ @ 7A@ @ @ 0 0  ( G}s=::c:Qy\:::r;cft|dkrw|d}t|tr|j}t|tjr|g}t|tst|}tj ddd|ifi|d|_ dS)Nr+rz celery.grouprUr-r&) r,rDr&r-rrUrrrrrrr8s r3rzgroup.__init__s u::??!HE%'' $ %!;<< (eV,, %e gu-=     #r;c|j|fi|Sr}r)rWrrs r3rzgroup.__call__st 88888r;c0t|||jS)N)r<r)r'rrs r3rz group.__or__"sT495555r;?ct|||d}|jD]%}|t|&|S)NT) repeatlast) countdown)r r-rnext)rWstartstopstepr{r9s r3skewz group.skew&sK UD$4 8 8 8J ) )D HHtBxxH ( ( ( ( r;Tc |r|nd}|td|td|j}|jjr|j||fi|S|js|S||\}} } ||jg| | |} t} t|j | ||| f||d|} |j | | | }| t|dkr#t|dtr|d}|j}|r|r|||S)NrUz%Cannot add link to group: use a chordz5Cannot add link to group: do that on individual tasksrr) ready_barrierr+r)r.rrFrGrr-r _freeze_gid _preparedrr0 _apply_tasksrfinalizer,rDcurrent_worker_task add_trail)rWrr add_to_parentrrrrrryrr-prQresult parent_tasks r3rzgroup.apply_async-s#tt  CDD D  !GII Ih 8 % 74:dF66g66 6z !;;== %)%5%5g%>%>"7tz2x#FF II(t(#qN.26NNELNNOO%%hq%II  v;;!   6!9k B B AYF-  *[ *  ! !& ) ) ) r;c rndrni|j}|js|S|\}}||jg|||}||fd|DS)NrUc8g|]\}}}|jddS)rrU)r)rrcrrrrs r3r3zgroup.apply..YsE* * * ?HsAqICI :4 : :' : :* * * r;)rr-rrrr)rWrrrrryrr-s ``` r3rz group.applyQs#tt!)rhz !;;== %)%5%5g%>%>"7tz2x#FFx* * * * * * LQ* * *    r;cD|jD]}||dSr}r-rrWrr9s r3rzgroup.set_immutable]s4J * *D   y ) ) ) ) * *r;Fc d}||j|fi|pi}|j||fi|}|j||fi|t|jt r.|jttf||d|ntg}|jD]<}t||j }|j ||fi|| |=t|jtr ||jdd<n||_||j|fi|dSdS)Nr>r?r)rXrrrDr-rmaprrAr)rr=rKrr[)rWr>r?r@r new_tasksr9s r3r=z group.stampasb  4g4TEEWEEKO%$%o}PPPP-;;7;;; dj& ) ) ' JNN8$5owVcoognoo p p p pI  ' '&t::: 7M==W===  &&&&$*o66 ' ) 111 &   G  1 1 1 1 1 1 1  r;c|d}|jd|S)NTrr)rrr-rrWrcs r3rz group.linkxs9iikkooo--z!}!!#&&&r;cDtfd|jDS)Nc3jK|]-}|dV.dS)TrN)rr)r child_taskrcs r3rz#group.link_error..s?cc*Z**399t9+D+DEEccccccr;)rr-rs `r3rzgroup.link_errors,ccccX\Xbccccccr;c #Kt|D]\} } || |r| } n || |} || tr'| | j||||} | Ed{Vk|r(| js!| || | jz| _| | ||| |fVdS)aRecursively unroll the group into a generator of its tasks. This is used by :meth:`apply_async` and :meth:`apply` to unroll the group into a list of tasks that can be evaluated. Note: This does not change the group itself, it only returns a generator of the tasks that the group would evaluate to. Arguments: tasks (list): List of tasks in the group (may contain nested groups). partial_args (list): List of arguments to be prepended to the arguments of each task. group_id (str): The group id of the group. root_id (str): The root id of the group. app (Celery): The Celery app instance. CallableSignature (class): The signature class of the group's tasks. from_dict (fun): Function to create a signature from a dict. isinstance (fun): Function to check if an object is an instance of a class. tuple (class): A tuple-like class. Returns: generator: A generator for the unrolled group tasks. The generator yields tuples of the form ``(task, AsyncResult, group_id)``. rN)ryrr) enumeraterr&rr-rrr) rWr-rryrrrUrrDrindexr9unrolls r3rzgroup._prepareds<%U++ i iKE4z$ 122 0zz||!y3///z$&& iJ h"!!!!!!!!GG %l 3 3eeDI6F6F FDIDKK7X]K^^`hhhhhh# i ir;c +K|p|j}||5}d} tj|\} }t | dt | d} t |D]\} }|\}}}||n|jd}| t |z } || |j || |j d |d|||d| |r5|j s.|js'|xjdz c_||dt | d} |V ddddS#1swxYwYdS) aURun all the tasks in the group. This is used by :meth:`apply_async` to run all the tasks in the group and return a generator of their results. Arguments: tasks (list): List of tasks in the group. producer (Producer): The producer to use to publish the tasks. app (Celery): The Celery app instance. p (barrier): Barrier object to synchronize the tasks results. args (list): List of arguments to be prepended to the arguments of each task. kwargs (dict): Dict of keyword arguments to be merged with the keyword arguments of each task. **options (dict): Options to be merged with the options of each task. Returns: generator: A generator for the AsyncResult of the tasks in the group. rNr'F)rrr'rrr+T)weakrU)rrrteerrrr_chord_descendrset_chord_sizer cancelledreadyr2then)rWr-rrrrr'rrr chord_size tasks_shifted next_task task_index current_taskrcrbry chord_objs r3rzgroup._apply_taskss0oTX  $ $X . .% ( J#,=#7#7 M5  % % %]D11I,5e,<,<  ( L &2"S(%*%6EECKOOGK..xDDD+&/d6++")++++Q[++FFaKFFHHQTH*** 55  7 % % % % % % % % % % % % % % % % % % sD D<<EEcijfd|D}|dtx|d<}|||dfS)z:Freeze the group id by the existing task_id or a new UUID.c@i|]\}}|jvs |jv||SrU)rr)rrrrWs r3rz%group._freeze_gid..sC& & & Q///1DL3H3H q3H3H3Hr;rryr)rrCr5rr)rWrrys` r3rzgroup._freeze_gids T\ & & & & $]]__& & &   KK 466 * * , h'++i"8"888r;c |j} |d}n%#t$r|p tx}|d<YnwxYw|r||d<|r||d<|||d<|d|}|d|}t |jt rjtj| |j\} } t| | ||||} t| |_nVg} t| | ||||} t |jtr | |jdd<n| |_|| fS)a?Freeze the tasks in the group. Note: If the group tasks are created from a generator, the tasks generator would not be exhausted, and the tasks would be frozen lazily. Returns: tuple: A tuple of the group id, and the AsyncResult of each of the group tasks. rryr'Nrrr)rrrrrDr-rrr _unroll_tasksr _freeze_tasksr0_freeze_unrollr) rWrryr'rrrrgidtasks1tasks2rQrs r3_freeze_group_taskszgroup._freeze_group_tasks s| 7y/CC 7 7 7$,$6 6C$y/// 7  ('D   "!DM  ""-D //)W55OOK;; dj& ) ) '']4+=+=dj+I+IJJNFFD..vxQZ[[\\GvDJJI4..8UGYG$*o66 ' ) 111 & G|rc R|jj|||||||S)N)rryr'rrr)rrr)rWrryr'rrrs r3rz group.freeze=s;#tx#T%=%=hI;&>& &   r;c#ZKfdt|DEd{VdS)zKCreates a generator for the AsyncResult of each task in the tasks argument.c3RK|]!\}}||V"dS)ryr'rrrN)r)rrr9r'ryrrs r3rz&group._freeze_tasks..Hs^?? * T KK&+(/*3,7  99??????r;N)r)rWr-ryr'rrs ````r3rzgroup._freeze_tasksFsv??????? .7u-=-= ??? ? ? ? ? ? ? ? ? ?r;c#4Kfd|DEd{VdS)z?Creates a generator for the cloned tasks of the tasks argument.c3hK|],}t|jV-dSr)r)rrrr9rWs r3rz&group._unroll_tasks..Rs>SSTODdi888>>@@SSSSSSr;NrUrWr-s` r3rzgroup._unroll_tasksOs<TSSSUSSSSSSSSSSSSr;c#Kt|j}d}|rt||j}t |tr||jn6| || |||||V|dz }|dSdS)aGenerator for the frozen flattened group tasks. Creates a flattened list of the tasks in the group, and freezes each task in the group. Nested groups will be recursively flattened. Exhausting the generator will create a new list of the flattened tasks in the group and will return it in the new_tasks argument. Arguments: new_tasks (list): The list to append the flattened tasks to. group_id (str): The group_id to use for the tasks. chord (Chord): The chord to use for the tasks. root_id (str): The root_id to use for the tasks. parent_id (str): The parent_id to use for the tasks. Yields: AsyncResult: The frozen task. rrrr+N) rr-r)popleftrrrDr& extendleftrKr) rWrryr'rrstackrr9s r3rzgroup._freeze_unrollTs*dj!!  !"5==?? BBBHHJJD$&& !  ,,,,  &&&kk8(-w,5.9";;;;;q  ! ! ! ! !r;ch|jr*t|jddd|jdSdS)Nrr9zgroup()zgroup())r-r rs r3rzgroup.__repr__xsD : *- 1 f%((((** * r;c*t|jSr})r,r-rs r3__len__z group.__len__s4:r;cr|j}|$ |jdj}n#t$rYnwxYw||ntSri)rr-rrjrrks r3rz group.appsQi ; jm'    oss;6rlr})rNr)NNTNNNr!r#)NNNNNNNr")%rsrtrurvrr-r$rrrrrrrrr=rrrrUr"rDrrrrrrr%rrrrrr&rr'r(s@r3r&r&s':  ^-> ? ?E5;5;5;[5;n $ $ $ $ $999666AE9=""""H     ***2222. ' ' 'ddd%-$>%/'u/i/i/i/ib>B/3'+>>>>@ 9 9 9BFFJ1111f599=    G???TTT "!"!"!H   77X77777r;r&r'c2eZdZdZeddZeddZ d fd ZddZ fdZ d!d Z d"d Z d#d Z d$dZedZdZ d%dZfdZdZdZdZdZedZddZeddZeddZxZS)&raLBarrier synchronization primitive. A chord consists of a header and a body. The header is a group of tasks that must complete before the callback is called. A chord is essentially a callback for a group of tasks. The body is applied with the return values of all the header tasks as a list. Example: The chord: .. code-block:: pycon >>> res = chord([add.s(2, 2), add.s(4, 4)])(sum_task.s()) is effectively :math:`\Sigma ((2 + 2) + (4 + 4))`: .. code-block:: pycon >>> res.get() 12 Ncp|}|jdi|d\}|d<||d|i|S)aCreate a chord signature from a dictionary that represents a chord. Example: >>> chord_dict = { "task": "celery.chord", "args": [], "kwargs": { "kwargs": {}, "header": [ { "task": "add", "args": [ 1, 2 ], "kwargs": {}, "options": {}, "subtask_type": None, "immutable": False }, { "task": "add", "args": [ 3, 4 ], "kwargs": {}, "options": {}, "subtask_type": None, "immutable": False } ], "body": { "task": "xsum", "args": [], "kwargs": {}, "options": {}, "subtask_type": None, "immutable": False } }, "options": {}, "subtask_type": "chord", "immutable": False } >>> chord_sig = chord.from_dict(chord_dict) Iterates over the given tasks in the dictionary and convert them to signatures. Chord header needs to be defined in d['kwargs']['header'] as a sequence of tasks. Chord body needs to be defined in d['kwargs']['body'] as a single task. The tasks themselves can be dictionaries or signatures (or both). rrrU)rr)rrrrrs r3rz_chord.from_dictsSp&&(("2#"2"G"GWX5F"G"GghsD-c-W---r;c ||f|fSr}rU)rgr<rs r3rz_chord._unpack_argss~v%%r; celery.chordc |r|nd}|r|ndii}tj||i|t||t||dfd|i|d|_dS)NrUrr)rgr<rr')rrrr)r) rWrgr<r9rrrrrs r3rz_chord.__init__s#tt!5"~tBFBl63.G.G"1$C"@"@"@BBB  GJ NU   $r;c ,|jd|rd|inifi|S)NrUr<r)rWr<rs r3rz_chord.__call__s,td$BVTNNNNgNNNr;ct|ttfs:t|tr%|}|j|z|_|St|trLt |jdkr4t|}|}|j|z|_|St |S)Nr+) rDr&rr"rr<r,r-r4rr)rWrrcrs r3rz _chord.__or__s55&/22 )5),, )**,,Cx%'CHJ u % % )#ek*:*:a*?*?'u--E**,,Cx%'CHJ77>>%(( (r;ct|jts t|j|j|_|j|||j}|jj|_d}|jr{|j|||||}|} t} | rJ| j| vrtd| | j| j || _ n | j } | J|S)Nr)rrr')rr'ryrzRecursive result parents) rDr-r&rrr<r r RuntimeErroraddrV) rWrryr'rrr header_result body_resultrcseens r3rz _chord.freeze s $*e,, 9tztx888DJ )) *CC *- 9 #)**WEH'+))K D55D #7d??&'ABBB!!!;&"/DK{ #r;Fc |j}t|tr|j}d}||j|fi|pi}|j||fi|}|j||fi|t|t r)|ttf||d|n(| }|D]}|j ||fi|||j |fi||<|j 7|j|fi|pi}|j||fi|}|j j ||fi|dSdSdS)Nr)r-rDr&rhrrrrrrArr=rjr<rl)rWr>r?r@r-rstampsr9s r3r=z _chord.stamp*s  eU # # KE  ;g;DLLGLLRPRO%$%o}PPPP-;;7;;; eV $ $ = IIh0j'Q^jjbijj k k k k\\^^F = = 7M< >g > > > > >  #8#8r;c 2|r|nd}|r|ni}|r+|js$t|t|jzn|j}|ddp |jd} t |jdfi|}| jdi| } || } t|j tr|j n;t|j | |j dt} | jjr8t!5|j||f| |d| cdddS#1swxYwY| rt |j fi| n|j } | dd}||}|j| | |f||d| S)NrUr<rr)rr)r<r)rr)rrrr5rrEr_get_apprDr-r&rrrrFrGrrrH)rWrrrrrLrrouter result_clsrr<rr-merged_optionsoption_task_ids r3rz_chord.apply_asyncDs6#tt!)r> $>d eDI....48I zz&$''>4;v+>dk(+66v66tz$$G$$mmD!!'1$*e'D'D^!!###DJC9I9I)UYU[U[9\9\]]]  8 % I"$$ I I!tz$I'+WII@GII I I I I I I I I I I I I I I I I;BSdl66g666t|'++It<< ?$GtxtT\76\\^\\\s'EE  E Tc J|r|nd}|r|ni}||jn|}t|jtr|jnt|j|j}|||||fS)NrUr) propagate)r)r<rDr-r&rrrr)rWrrrr<rr-s r3rz _chord.apply^s#tt!)r Ltyyd'1$*e'D'D6!!###DJDH555 zz++dF++//)/DDF   r;c^t|ts/t|trt|}t|tr6t |jd|j}tfd|DSt|tr4|jdddD]!} |}|dkr|cS"dSt|tr |j St|trdSt|S)zCount the number of tasks in the given signature recursively. Descend into the signature object and return the amount of tasks it contains. r-c3BK|]}|VdSr}r)rr9rs r3rz"_chord._descend..us/??ds||D))??????r;r;Nrr+) rDr"rErr&r7r-sumrrr'r<r,)rsig_objsubtasks child_sig child_sizes` r3rz_chord._descendis.'9-- 3*Wd2K2K 3))'22G gu % % w}gw}EEH????h????? ?  ( ( $]26r62 & &  \\)44 >>%%%%"1  ' ' << -- -  + + 17||r;cptjdj}tfd|DS)z@Return the number of tasks in this chord's header (recursively).r-c3BK|]}|VdSr}rrs r3rz)_chord.__length_hint__..s/9944==&&999999r;)r7r-rrs` r3r/z_chord.__length_hint__s; GTZ8899995999999r;r+c  |p||}|jdp t} |jd} | rt |jfi| n|j} | r0| dd|j| || | }| dd| dd| dd|| || }|j r1|j ||||||j || fd| i| }n$| g|jj|}||_|S) aExecute the chord. Executing the chord means executing the header and sending the result to the body. In case of an empty header, the body is executed immediately. Arguments: header (group): The header to execute. body (Signature): The body to execute. partial_args (tuple): Arguments to pass to the header. app (Celery): The Celery app instance. interval (float): The interval between retries. countdown (int): The countdown between retries. max_retries (int): The maximum number of retries. task_id (str): The task id to use for the body. kwargs (dict): Keyword arguments to pass to the header. options (dict): Options to pass to the header. Returns: AsyncResult: The result of the body (with the result of the header in the parent of the body). rrNrTr#r')ryr'r)intervalr max_retries)rrrrrEr5rrrr-r apply_chordrrrrrV)rWrgr<rrr rr!eagerrrrryrbodyresheader_result_argsrs r3rHz _chord.runs0(T]]4((>%%i00:DFF,""9--3:L$t|//w///   ) KK 4 ( ( ( L   ( ( (++gw+77  GT""" GT""" It$$$#77QU_f7gg < F K # #"!#' $   /F.|VaaXaY`aaMM JJrNNN0DH02DEM&r;ctj|i|} t|jdd|jd<n#tt f$rYnwxYw|S)Nr<TrB)rrr)rr1r)rWrrr(rs r3rz _chord.clonesy!EGGM42622  '6 ((6(6(6I V $ $)    D s$AAAc:|j||S)z(Links a callback to the chord body only.)r<rrs r3rz _chord.links x   r;c|jjjrCt|jpgD]+}||d,ntjdt|j ||S)atLinks an error callback to the chord body, and potentially the header as well. Note: The ``task_allow_error_cb_on_chord_header`` setting controls whether error callbacks are allowed on the header. If this setting is ``False`` (the current default), then the error callback will only be applied to the body. TraKtask_allow_error_cb_on_chord_header=False is pending deprecation in a future release of Celery. Please test the new behavior by setting task_allow_error_cb_on_chord_header to True and report any concerns you might have in our issue tracker before we make a final decision regarding how errbacks should behave when used with chords.) rrF#task_allow_error_cb_on_chord_headerrr-rrwarningswarnrr<)rWrqr9s r3rz_chord.link_errors 8= < "4:..4" ? ?  = =>>>> ? MN +     W%%%r;cD|jD]}||dS)zSets the immutable flag on the chord header only. Note: Does not affect the chord body. Arguments: immutable (bool): The new mutability value for chord header. Nrrs r3rz_chord.set_immutables6J * *D   y ) ) ) ) * *r;c |jrt|jtrt|jjddd|jjd|jt|jjdd|jSdt|jd|j|jzSd|jdS) Nrr9z %({} | {!r})r+r%z) r<rDrr r-formatr r#rrs r3rz_chord.__repr__ s 9 C$)V,, 1IOA&v.")) *33DJ??dioabb1tyAAA3 &!49#5#5dj#A#ACCC C6tz6666r;c6||jSr})rr<rs r3rz _chord.app s}}TY'''r;c|j}|? |jj}n#t$r |j}YnwxYw|r |dj}| ||j}||ntSri)rr-r1r)rWr<rr-s r3rz_chord._get_app swi ; # (! # # #  # $Ahm{t/ioss;6s ,,z kwargs.headerzTasks in chord header.z kwargs.bodyzBody task of chord.r}r!)NrNNNr"r#)NNNNNNNN)NNTN)NNr+NFNN)rsrtrurvr$r staticmethodrrrrrr=rrrr/rHrrrrrrrrrr-r<r'r(s@r3rrs49.9.9.[9.v&&&\& 0>-1$$$$$$OOOO)))))"599=B????4;?>B,0]]]]4'+#'     [<::: BF16!%;;;;z < * * * 7 7 7((_( 7 7 7 7  _.F G GE  M+@ A ADDDDDr;rc |d}t|trJt|tjr|St ||St |g|Ri|S)zCreate new signature. - if the first argument is a signature already then it's cloned. - if the first argument is a dict, then a Signature version is returned. Returns: Signature: The resulting signature. rr)rrDrErrUrr"r)variesrrrs r3r(r(+ s **U  C&$4 fh8 9 9 "<<>> !""6s"333 V -d - - -f - --r;Fc|^t|tjr|r|}n$t|trt |}|||_|S)aEnsure obj is a signature, or None. Arguments: d (Optional[Union[abstract.CallableSignature, Mapping]]): Signature or dict-serialized signature. app (celery.Celery): App to bind signature to. clone (bool): If d' is already a signature, the signature will be cloned when this flag is enabled. Returns: Optional[abstract.CallableSignature] )rDrrUrrEr(r)rrrs r3r)r)? sb } a3 4 4  GGII 4  ! A ?AF Hr;)Tr#)Orvrrur*abcrr collectionsrcollections.abcrrr functoolsrrr r typesr kombu.utils.functionalr r kombu.utils.objectsrkombu.utils.uuidrviner celery._statercelery.exceptionsr celery.resultrr celery.utilsrcelery.utils.collectionsrcelery.utils.functionalrrrrrrrrcelery.utils.objectsrcelery.utils.textr r!__all__r4r:rArFrRrUregisterrEr"r+rrr#rzr$r%rr&rr'r(subtaskr) maybe_subtaskrUr;r3rLs;''''''''++++++))))))44444444//////!!!!!!%%%%%%88888888888888!!!!!!------******555555______________111111BBBBBBBB  666$'''****Znnnnnnnnnb $ D D D D D D D%$ DD6!6!6!r g&&g5g5g5g5g5Yg5g5'&g5T <6<6<6<6<6F<6<6<6~     y   0  F F F F F8 F F F GGGGGxGGG #+#+#+#+#+Y#+#+#+L     R7R7R7R7R7IR7R7R7j g&&UBUBUBUBUBYUBUB'&UBt ..."     6 r;