o wvXj@sdZddlmZmZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZdd lmZmZdd lmZdd lmZdd lmZdd lmZddlmZmZddlmZddlm Z m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'ddl&m(Z)ddl&m*Z*m+Z+m,Z,m-Z-m.Z.ddl/m0Z0ddl1m2Z2m3Z3zddl4m5Z5Wne6yddlm5Z5YnwdZ7eoed Z8ddZ9ddZ:dd Z;e#jZ?d#d$Z@e?jAd%d&eGd'd(d(e?ZBGd)d%d%eBZCGd*d+d+e?ZDe?AeGd,d-d-eDZEe?AeGd.d/d/eDZFe?AGd0d1d1e?Z(d2d3ZGe?AeGd4d5d5e?ZHe?AeGd6d7d7e?ZId8d9ZJeJZKd=d;d<ZLeLZMdS)>zmComposing task work-flows. .. seealso: You should import these from :mod:`celery` and not this module. )absolute_importunicode_literalsN)deque)deepcopy)partial)reduce) itemgetter)fxrangereprcall)cached_property)uuid)barrier) current_app)PY3python_2_unicode_compatible) try_import) GroupResultallow_join_result)abstractChainMap)_regen)chunks)is_list maybe_listregenseq_concat_itemseq_concat_seq)getitem_property)remove_repeating_from_tasktruncate)MutableSequence) Signaturechainxmapxstarmaprgroupchord signaturemaybe_signature simplejsonc Cszt|j}Wn/ty6z|j}Wnttfy$|YYSw|dkr2t|jdYS|YSw|dkr@|jdS|S)z"Unroll group with only one member.r)lentasks TypeError__length_hint__AttributeErrorlist)r&sizer3J/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/canvas.pymaybe_unroll_group3s  "r5cCs t|d|S)Nname)getattrtaskr3r3r4task_name_fromC r:cCs|j|dd|S)z?Used by custom signatures in .from_dict, to keep common fields. chord_size)r<)updateget)fieldssigr3r3r4_upgradeGsrAcseZdZdZiZdZZed\ddZed\ddZ   d]fdd Z d d Z d d Z d^ddZ d_ddZd`ddZd^ddZeZ  daddZeZd_ddZd\ddZddZddZd d!Zd"d#Zd$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Z d2d3Z!d4d5Z"d6d7Z#d8d9Z$d:d;Z%e&rdd?Z)e*d@dAZ+e*dBdCZ,e*dDdEZ-e*dFdGZ.e/dHdIZ0e/dJdKZ1e/dLdMZ2e/dNdOZ3e/dPdQZ4e/dRdSZ5e/dTdUZ6e/dVdWZ7e/dXdYZ8e/dZd[Z9Z:S)br"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 chaning `.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={}} Ncsfdd}|S)Ncs|jp|j<|SN)TYPES__name__)subclassclsr6r3r4_innersz'Signature.register_type.._innerr3)rGr6rHr3rFr4 register_typeszSignature.register_typecCs:|d}|r|j|}||ur|j||dSt||dS)N subtask_typeapp)r>rC from_dictr")rGdrLtyp target_clsr3r3r4rMs   zSignature.from_dictFc s||_t|trtt||dSz|j} Wn ty"|} Ynw||_tt|j| t |p0d|p4it|p8ifi| ||dddS)Nr3)r9argskwargsoptionsrJ immutabler<) _app isinstancedictsuperr"__init__r6r0_typetuple) selfr9rQrRrStyperJrTrLex task_name __class__r3r4rYs"      zSignature.__init__cOs$|||d\}}}|j|i|S)z0Call the task directly (in the current process).N)_merger])r\ partial_argspartial_kwargsrQrR_r3r3r4__call__szSignature.__call__cOs |||S)z5Shortcut to :meth:`apply_async` using star arguments. apply_async)r\rcrdr3r3r4delay zSignature.delaycKsT|r|nd}|r |ni}dd|D}||||\}}}|jj||fi|S)zCall task locally. Same as :meth:`apply_async` but executed the task inline instead of sending a task message. r3cSi|] \}}|dur||qSrBr3.0kvr3r3r4 z#Signature.apply..)itemsrbr]apply)r\rQrRrSr3r3r4rss  zSignature.applycKs|r|nd}|r |ni}dd|D}z|j}Wn ty$YdSw|s+|s+|r6||||\}}}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. r3cSrkrBr3rlr3r3r4rprqz)Signature.apply_async..N)rr _apply_async IndexErrorrbrQrRrS)r\rQrR route_namerS_applyr3r3r4rhs     zSignature.apply_asynccCs|r|nd}|r |ni}|r|ni}|jr+|s+|j|j|r't|jfi|fS|jfS|r6t|t|jn|j|rCt|jfi|n|j|rQt|jfi|fS|jfSNr3)rTrQrRrWrSr[)r\rQrRrSforcer3r3r4rbs$     zSignature._mergec Ks|r|nd}|r |ni}|s|s|r||||\}}}n |j|j|j}}}tj|jt||t||j |j |j d|j d}|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. r3)r9rQrRrSrJr<rTrK)rbrQrRrSr"rMr9r[rrJr<rTrUrZ)r\rQrRoptsr(r3r3r4clones"  zSignature.clonecCs|j}z|d}Wnty|pt}|d<Ynw|r"||d<|r(||d<d|vr2|jj|d<|r8||d<|r>||d<||S)a6Finalize the signature by adding a concrete task id. The task won't be called and you shouldn't call the signature twice after freezing it as that'll result in two task messages using the same task id. Returns: ~@AsyncResult: promise of future evaluation. task_idroot_id parent_idreply_togroup_idr')rSKeyErrorr rLoid AsyncResult)r\_idrr'r}r~rztidr3r3r4freezes"     zSignature.freezecCs6|}|dur ||_|dur||_|dur||_|S)zReplace the args, kwargs or options set for this signature. These are only replaced if the argument for the section is not :const:`None`. N)r{rQrRrS)r\rQrRrSr(r3r3r4replace8szSignature.replacecKs"|dur |||j||S)uSet arbitrary execution options (same as ``.options.update(…)``). Returns: Signature: This is a chaining method call (i.e., it will return ``self``). N) set_immutablerSr=)r\rTrSr3r3r4setGs  z Signature.setcCs ||_dSrBrT)r\rTr3r3r4rS zSignature.set_immutablecCs,|j|g}t|ts|g}|j|<|SrB)rS setdefaultrVr!)r\keyrrr3r3r4_with_list_optionVs zSignature._with_list_optioncCs ||}||vr|||SrB)rappendr\rvaluerrr3r3r4append_to_list_option\s  zSignature.append_to_list_optioncCs||}|t|dSrB)rextendrrr3r3r4extend_list_optionbs zSignature.extend_list_optioncC |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`. linkrr\callbackr3r3r4rf zSignature.linkcCr)zAdd callback task to be applied on error in task execution. Returns: Signature: the argument passed, for chaining or use with :func:`~functools.reduce`. link_errorrr\errbackr3r3r4rorzSignature.link_errorcCs|||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`. )rrr3r3r4on_errorxs zSignature.on_errorc Cs6ttjt|ggddt|jdpgDS)zhReturn a recursive list of dependencies. "unchain" if you will, but with links intact. css|]}|VqdSrB) flatten_links)rmrr3r3r4 sz*Signature.flatten_links..r)r1 itertoolsr# from_iterablerrSr>r\r3r3r4rs zSignature.flatten_linkscCst|tr t|||jdSt|tr5t|}t|tr-|}|s#|Stt|||jdSt|||jdSt|tsLt|trLtt |f||jdSt|tr_tt |||jdSt|trp| }|j |B|_ |St|t rt|tr|j rt|j dtr| }t|j d||jd|j d<|S|j rt|j dtr| }|j dj |B|j d_ |Stt|||jdSt|||jdStS)N)bodyrLrK)rVr&r'rUr5_chain unchain_tasksrrLrr{rr"r-NotImplemented)r\otherr-r@r3r3r4__or__sb           zSignature.__or__cCs|j}|j}|jdpt}|d'}|j||}|jj |d|j dd|i||j d| |WdS1s>wYdS)Nr|r9) connectionr3) r]rLrSr>r producer_or_acquirebackend on_task_callcontrolelectionr{rr)r\r]rLrproducerpropsr3r3r4rs  $zSignature.electioncOs(|j||idd\}}}t|d||S)NT)ryr9)rbr )r\rQrRrer3r3r4r szSignature.reprcallcCs||t|<t|SrB)idrW)r\memor3r3r4 __deepcopy__s zSignature.__deepcopy__cCs |SrB)rhr>rr3r3r4 __invert__r;zSignature.__invert__cCstt|ffSrB)r(rWrr3r3r4 __reduce__szSignature.__reduce__cCst|SrB)rWrr3r3r4__json__zSignature.__json__cCs|SrB)r rr3r3r4__repr__rzSignature.__repr__ccs6t|D]\}}t|tr|n||fVqdSrB)rWrrrVbytesdecode)r\rnror3r3r4rrszSignature.itemscCs|jSrBr8rr3r3r4r6szSignature.namecCs|jp |jj|dSNr9)rZrLr-rr3r3r4r]szSignature.typecCs |jptSrB)rUrrr3r3r4rLs z Signature.appcCs&z|jjWSty|jjYSwrB)r]rrrLrr3r3r4rs    zSignature.AsyncResultcCs0z|jjWStyt|jj|dYSwr)r]rhr_partialrL send_taskrr3r3r4rts   zSignature._apply_asynczoptions.task_idz Task UUIDzoptions.parent_idzTask parent UUID.zoptions.root_idzTask root UUID.r9z Name of task.rQzPositional arguments to task.rRzKeyword arguments to task.rSzTask execution options.rJzType of signaturer<zSize of chord (if applicable)rTz+Flag set if no longer accepts new argumentsrB)NNNNNNFNNN)NNN)NNNFNNNNN);rD __module__ __qualname____doc__rCrUrZ classmethodrIrMrYrfrirsrhrbr{rr_freezerrrrrrrrrrrrr rrrrrJSON_NEEDS_UNICODE_KEYSrrpropertyr6r r]rLrrtrrr~r}r9rQrRrSrJr<rT __classcell__r3r3r`r4r"Ms8     !         6               r"cCsJ|r tddi|Sd|vrtd|i|S|dur#td|d|i|SdS)Nr#r)rSr-use_linkr3r3r4_prepare_chain_from_optionss)rr#)r6c @seZdZeddZedddZddZdd Zd d Z d d Z dddZ   dddZ  d ddZ ddddddddejf ddZdddZeddZddZdS)!r kwargs.taskszTasks in chain.Ncs^|dd}|r t|trt|}|dd<fdd|D}t|t|fdi|dS)NrRr-cg|]}t|dqSrK)r))rmr9rKr3r4 Zz$_chain.from_dict..rLrS)rVr[r1rAr)rGrNrLr-r3rKr4rMTs  z_chain.from_dictcOsbt|dkrt|drt|dn|}tj|ddd|ifi||dd|_d|_d|_dS)Nr+rz celery.chainr3r-rr#) r,rrr"rYpop _use_linkrJ_frozenr\r-rSr3r3r4rY]s$  z_chain.__init__cOs|jr |||SdSrB)r-rh)r\rQrRr3r3r4rfgs z_chain.__call__cs@ttjg|Ri|}fdd|jdD|jd<|S)Ncsg|] }|jddqS)T)rLr{)rU)rmr@r\ to_signaturer3r4rnsz _chain.clone..r-)r)r"r{rRr\rQrRr(r3rr4r{ks   z _chain.clonecCs<dd|jD}|jdgD] }|D]}||qq|S)NcSsg|]}|qSr3r{rmtr3r3r4rwsz(_chain.unchain_tasks..r)r-rSr>r)r\r-r@r9r3r3r4rts  z_chain.unchain_taskscKs|r|nd}|r |ng}|j}|jjr0t|j||fi|WdS1s+wY|j||fd|i|rEt|jfi|S|jS)Nr3rL)rLconftask_always_eagerrrsrunrWrS)r\rQrRrSrLr3r3r4rh}s   z_chain.apply_asyncc  Ks|r|nd}|r |ng}| p|j} |j}|dur | jjdkr d}|r.|js.t|t|jn|j}||||j| | || ||| \}}|rd|rN|d d|| }t | ||} |j di| |dSdS)Nr3r+Trr) rLrr task_protocolrTr[rQ prepare_stepsr-rrrrh)r\rQrRrr'r|rr publisherrr}r~rLrSrr-results first_taskr3r3r4rs0    z _chain.runc Cs:|j|j|j|j||d|j|||dd \}}|_|dS)NFrr)rrQrRr-rLr)r\rrr'r}r~rerr3r3r4rs  z _chain.freezeTc  Cs|p|j}|j} | dur|jjdkrd} t|}|j}|j}d}d}gg}}d}|r|}| | }}t|tj sA| ||d}t|t rJt |}| rZ|rU| ||}n| }n |rft |t |j|_t|trq||jq)t|t r|r||z t|||j||d}Wntyt||||d}Ynw|r|j||| | d}n|j|d}|d7}|r| r|||r|js||_|rt|D]}||q||||||}}t|tr|j|}|jr|j}|js|}|s+||fS) Nr+TrrK)rr|r}rL)rr}rL)r}rr'r})rLrrrrrrrVrCallableSignaturer&r5r{r[rQrr-r'r|r0rrparentrrrrensure_chords_allowed)r\rQrRr-r}r~rrL last_task_idr chord_bodyr{rMrsteps steps_pop steps_extend prev_taskprev_resrir9 is_first_task is_last_taskresrnoder3r3r4rs                     Wz_chain.prepare_stepsc Ks|r|nd}|r |ni}d||f}\}}|jD]&}|||j|o&|ffit|jfi|}||d|_}\}}q|S)Nr3r)r-r{rsr>rWrSr) r\rQrRrSlastfargsfkwargsr9rr3r3r4rs"s     z _chain.applycCs:|j}|durz|jdj}Wn tyYnw|ptSNr)rUr- LookupErrorrr\rLr3r3r4rL,s z _chain.appcCsB|jsdt|jt|St|jddddd|jDS)Nz<{0}@{1:#x}: empty>rr9z | css|]}t|VqdSrB)reprrr3r3r4r<sz"_chain.__repr__..)r-formatr]rDrrjoinrr3r3r4r6s z_chain.__repr__rBr) NNNNNNNNNNNNr)rDrrrr-rrMrYrfr{rrhrrr"rrsrrLrr3r3r3r4rOs0         t  rcs eZdZdZfddZZS)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. csf|s$|r$t|dkst|dr$t|dkr|dn|}ttj|tStt|j|g|Ri|S)Nr+r)r,rroperatoror_r#rX__new__)rGr-rRr`r3r4rps z chain.__new__)rDrrrrrr3r3r`r4r#?s0c@s:eZdZdZeddZed ddZddZd dd Z dS) _basemapNr9itcCs&t||||dd|i|dSNrRrLrS)rA _unpack_argsrGrNrLr3r3r4rM z_basemap.from_dictcKs,tj||jd|t|dfddi|dS)Nr3r9rrTT)r"rY _task_namer)r\r9rrSr3r3r4rYs  z_basemap.__init__cKsX|r|nd}|r |ni}||j\}}|jjd|t|dfdt|jdi|S)Nr3rrvr9)rrRr]rhr1r:r>)r\rQrRrzr9rr3r3r4rhs  z_basemap.apply_asyncrBr) rDrrrrrrrMrYrhr3r3r3r4r{s  rc@eZdZdZdZddZdS)r$zMap operation for tasks. Note: Tasks executed sequentially in process, this is not a parallel operation like :class:`group`. z celery.mapcC(||j\}}d|jtt|dS)Nz[{0}(x) for x in {1}]drrRrr9r rr\r9rr3r3r4rz xmap.__repr__NrDrrrrrr3r3r3r4r$s r$c@r)r%z.Map operation for tasks, using star arguments.zcelery.starmapcCr)Nz[{0}(*x) for x in {1}]r r r r3r3r4rr zxstarmap.__repr__Nr r3r3r3r4r%s r%c@sZeZdZdZedddZedddZdd Zd d Z dd d Z ddZ edddZ dS)rz)Partition of tasks into chunks of size n.r9rnNcCs&t|t||dd|i|dSr)rArrrr3r3r4rMszchunks.from_dictcKs,tj|dd|t||dfddi|dS)Nz celery.chunksr3)r9rrrTT)r"rYr)r\r9rrrSr3r3r4rYs zchunks.__init__cKs|jdi|Srxrg)r\rSr3r3r4rfszchunks.__call__cKs@|r|nd}|r |ni}|j||fdt|jdi|S)Nr3rvr9)r&rhr:rRr>)r\rQrRrzr3r3r4rhs  zchunks.apply_asynccs:j\}}tfddtt||DjdS)Nc3s |] }t|jdVqdS)rKN)r%rU)rmpartr\r9r3r4rszchunks.group..rK)rrRr&_chunksiterrU)r\rrr3rr4r&s  z chunks.groupcCs|||||dS)NrKr3)rGr9rrrLr3r3r4 apply_chunksszchunks.apply_chunksrBr) rDrrrrrrrMrYrfrhr&rr3r3r3r4rs   rcsZt|tr t|d}t|ttfr|j}|St|tjr"|g}|Sfdd|D}|S)NrKcrr)r(rrKr3r4rrz _maybe_group..)rVrWr(r&rr-rr)r-rLr3rKr4 _maybe_groups   rc@seZdZdZeddZed)ddZddZd d Z d*d d Z  d+ddZ d,ddZ ddZ ddZddZejejeefddZ   d-ddZddZ  d.dd ZeZd!d"Zd#d$Zd%d&Zed'd(ZdS)/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). rzTasks in group.NcCs&t|t|ddfd|i|dS)NrRr-rLrS)rAr&rr3r3r4rMrzgroup.from_dictcOsrt|dkr&|d}t|tr|j}t|tjr|g}t|ts&t|}t j |ddd|ifi|d|_ dS)Nr+rz celery.groupr3r-r&) r,rVr&r-rrr{rrr"rYrJrr3r3r4rYs       zgroup.__init__cOs|j|fi|SrBrg)r\rcrSr3r3r4rf#szgroup.__call__?cCs0t|||dd}|jD] }|jt|dq |S)NT) repeatlast) countdown)r r-rnext)r\startstopsteprr9r3r3r4skew&s z group.skewTcKs|r|nd}|durtd|durtd|j}|jjr'|j||fi|S|js.|S||\}} } ||jg| | |} t } t |j | ||| f||d|} |jj | | | d}| t|dkrrt|dt rr|d}|j}|r~|r~|||S)Nr3z%Cannot add link to group: use a chordz5Cannot add link to group: do that on individual tasksrQrR) ready_barrierr+r)r.rLrrrsr-r _freeze_gid _preparedr r1 _apply_tasksrfinalizer,rVcurrent_worker_task add_trail)r\rQrR add_to_parentrrrrSrLrr}r-prresult parent_taskr3r3r4rh,s8  zgroup.apply_asyncc snrndr ni|j}|js|S|\}}||jg|||}||fdd|DS)Nr3cs&g|]\}}|jddqS)rr3)rs)rmr@rerQrRrSr3r4rXszgroup.apply..)rLr-rrr r)r\rQrRrSrLrr}r-r3r)r4rsPs  z group.applycC|jD]}||qdSrBr-rr\rTr9r3r3r4r\s  zgroup.set_immutablecCs |jdd}|jd|S)NTrr)r{rr-rr\r@r3r3r4r`sz group.linkcCsJz |jdd}Wntyd|d<t|}Ynw|jd|S)NTrrTr)r{rr0r"rMr-rr-r3r3r4res zgroup.link_errorc cs|D]G} || |r| } n|| |d} || tr1| | j||||} | D] \} } | | fVq&q|r@| js@| || | j| _| | j||dfVqdS)NrK)rr})r{r&r r-rTrQr)r\r-rcrr}rLrrMrVr[r9unrolltaskNresNr3r3r4r os         zgroup._preparedc ks|p|j}||=}|D]1\} } | jd|d| jdp|||d| |r<|js<|js<|jd7_| j|dd| VqWddS1sKwYdS)NFr')rr%r'rQrRr+T)weakr3) rLrrhrSr> cancelledreadyr2then) r\r-rrLr&r%r'rQrRrSr@rr3r3r4r!s      "zgroup._apply_taskscCs8t|jfi|}|dt|d<}|||dfS)Nr|rr})rWrSrr r>)r\rSrr3r3r4rs  zgroup._freeze_gidc Cs|j}z|d}Wnty|pt}|d<Ynw|r"||d<|r(||d<|d|}|d|}g}t||||||} t|jtrO||jdd<n||_|j || S)Nr|rr'r}r~) rSrr rr1_freeze_unrollrVr-r!rLr) r\rrr'r}r~rzgid new_tasksrr3r3r4rs(      z group.freezeccsht|j}|r2t||jd}t|tr||jn| ||j ||||dV|sdSdS)NrK)rr'r}r~) rr-r)popleftrUr{rVr& extendleftrr)r\r7rr'r}r~stackr9r3r3r4r5s    zgroup._freeze_unrollcCs$|jrt|jddd|SdS)Nrr9zgroup({0.tasks!r})zgroup())r-rrrr3r3r4rs  zgroup.__repr__cCs t|jSrB)r,r-rr3r3r4__len__rz group.__len__cCsB|j}|durz|jdj}Wn tyYnw|dur|StSr)rUr-rLrrrr3r3r4rLs z group.apprB)rNr)NNTNNNr)NNNNNNNr)rDrrrrr-rrMrYrfrrhrsrrrrrr"rVr[r r!rrrr5rr;rrLr3r3r3r4r&sB    $     r&c@seZdZdZed-ddZed.ddZ  d/dd Zd-d d Z  d0d d Z   d1ddZ  d2ddZ d-ddZ ddZ   d3ddZddZddZdd Zd!d"Zd#d$Zed%d&Zd-d'd(Zed)d*Zed+d,ZdS)4r'aLBarrier 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 NcCs<|}|jdi|d\}|d<t|||d|i|S)NrRrLr3)copyrrA)rGrNrLrSrQr3r3r4rMszchord.from_dictcKs ||f|fSrBr3)headerrrRr3r3r4r s zchord._unpack_args celery.chordc KsT|r|nd}|r |ni}tj||||t||t||ddfd|i|d|_dS)Nr3rK)rRr=rrLr')r"rYrr)rJ)r\r=rr9rQrRrLrSr3r3r4rYs     zchord.__init__cKs |jd|r d|inifi|S)Nr3rrg)r\rrSr3r3r4rfs zchord.__call__c Cst|jtst|j|jd|_|jj|||jd}|jj||||d}|}t} |rG|j| vr3td| |j|j durB||_ n|j }|s*|jj|_|S)NrK)r~r}r')r}r'rzRecursive result parents) rVr-r&rLrrrr RuntimeErroraddr) r\rrr'r}r~ header_result body_resultrseenr3r3r4rs*     z chord.freezec Ks|r|nd}|r |ni}|r|jst|t|jn|j}|ddp'|jd} t|jdfi|}| jdi| } || } t|j t rK|j nt |j | d} | j j rvt |j||f| |d| WdS1sqwY|j| | |fd|i| S)Nr3rrRrK)rr|r|)rTr[rQrrRrWr{_get_apprVr-r&rrrrsr) r\rQrRr|rrrrouter result_clsrSrrLr-r3r3r4rh:s0      zchord.apply_asyncTcKsn|r|nd}|r |ni}|dur|jn|}t|jtr |jnt|j|jd}|j|||j|dfdS)Nr3rK) propagate)rQ)rrVr-r&r{rLrsr>)r\rQrRrGrrSr-r3r3r4rsNs  z chord.applyccszt|}|r;|}t|tr||jn t|tr.t|jdtr.||jdjn |dur4|n|V|sdSdS)Nr)rr8rVr&rr-r)r\r-rr:r9r3r3r4_traverse_tasksYs  zchord._traverse_taskscCs*t|jtr |jjn|j}t||dS)Nr+)rVr-r&sumrH)r\r-r3r3r4r/dszchord.__length_hint__r+Fc Ks|p||}|jdpt} |jd} ||_| r&t|jfi| n|j} | r7| dd|j| |j | | d} | dd| dd| dd|j | || d}t |dkrs|j j |||||d||d| i| }n| g|| _| S) Nr|r}rr#r')rr'r}r)intervalr max_retries)rDrSr>r r/r<rWrr=rr,r apply_chordrir)r\r=rrcrLrJrrKeagerr|rSrr}bodyresrAr3r3r4ris2         z chord.runc OsPtj|g|Ri|}zt|jddd|jd<W|Sttfy'Y|Sw)NrTr)r"r{r)rRr0rrr3r3r4r{s z chord.clonecC|j||SrB)rrrr3r3r4r z chord.linkcCrOrB)rrrr3r3r4rrPzchord.link_errorcCr*rBr+r,r3r3r4rs  zchord.set_immutablec Cs~|jr:t|jtr+t|jjddd|jjd|jt|jjdd|jdSdt|jd|j|jSd|S)Nrr9z%({0} | {1!r})r+rK%z!) rrVrrr-rr r#rUrr3r3r4rs  zchord.__repr__cCs ||jSrB)rDrrr3r3r4rLrjz chord.appcCsl|j}|dur.z|jj}Wn ty|j}Ynwt|r#|dj}|dur.|dur.|j}|dur4|StSr)rUr-r0r,r)r\rrLr-r3r3r4rDs    zchord._get_appz kwargs.headerzTasks in chord header.z kwargs.bodyzBody task of chord.rBr)Nr>NNNr)NNNNNNNN)NNTN)NNr+NFN)rDrrrrrM staticmethodrrYrfrrhrsrHr/rr{rrrrr rLrDrr-rr3r3r3r4r'sH        (    r'cOsL|d}t|trt|tjr|Stj||dSt|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. rLrK)r>rVrWrrr{r"rM)variesrQrRrLr3r3r4r(s  r(FcCsF|dur!t|tjr|r|}n t|trt|}|dur!||_|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] N)rVrrr{rWr(rU)rNrLr{r3r3r4r)s  r))NF)Nr __future__rrrr collectionsrr<r functoolsrrrrkombu.utils.functionalr r kombu.utils.objectsr kombu.utils.uuidr viner celery._stater celery.fiverr celery.localr celery.resultrr celery.utilsrcelery.utils.collectionsrcelery.utils.functionalrrrrrrrrcelery.utils.objectsrcelery.utils.textrr collections.abcr! ImportError__all__rr5r:rArregisterrWr"rrIrr#rr$r%rr&r'r(subtaskr) maybe_subtaskr3r3r3r4s                J 9o< * yc