U d7@s:dZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z ddlm Z dd lmZdd lmZmZdd lmZdd lmZdd lmZddlmZddlmZmZddlmZddl m!Z!ddl"m#Z#ddl"m$Z%ddl"m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+ddl,m-Z-ddl.m/Z/m0Z0dZ1ddZ2ddZ3ej4j5Gddde6Z7ddZ8e7j9d d!Gd"d#d#e7Z:Gd$d d e:Z;Gd%d&d&e7Ze79Gd+d,d,e7Z$d-d.Z?e79Gd/d0d0e7Z@e7j9d1d!Gd2d3d3e7ZAeAZBd4d5ZCeCZDd9d7d8ZEeEZFdS):zmComposing task work-flows. .. seealso: You should import these from :mod:`celery` and not this module. N)deque)MutableSequence)deepcopy)partial)reduce) itemgetter) GeneratorType)fxrangereprcall)cached_property)uuid)barrier) current_app) GroupResultallow_join_result)abstractChainMap)_regen)chunks)is_list lookahead maybe_listregenseq_concat_itemseq_concat_seq)getitem_property)remove_repeating_from_tasktruncate) Signaturechainxmapxstarmaprgroupchord signaturemaybe_signaturec Cszt|j}Wnbtk rpz|j}Wn ttfk rL|YYSX|dkrdt|jdn|YSYnX|dkr|jdS|SdS)z"Unroll group with only one member.rN)lentasks TypeError__length_hint__AttributeErrorlist)r#sizer/1/tmp/pip-unpacked-wheel-9cz4377o/celery/canvas.pymaybe_unroll_group(s$r1cCs t|d|S)Nname)getattrtaskr/r/r0task_name_from8sr6cseZdZdZiZdZZdhZed]ddZ ed^ddZ d_fd d Z d d Z d dZ d`ddZdaddZdbddZdcddZeZddddZeZdeddZdfddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zd/d0Z d1d2Z!d3d4Z"d5d6Z#d7d8Z$d9d:Z%d;d<Z&d=d>Z'fd?d@Z(e)dAdBZ*e+dCdDZ,e+dEdFZ-e+dGdHZ.e+dIdJZ/e0dKdLZ1e0dMdNZ2e0dOdPZ3e0dQdRZ4e0dSdTZ5e0dUdVZ6e0dWdXZ7e0dYdZZ8e0d[d\Z9Z:S)gra.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={}} Ngroup_idcsfdd}|S)Ncs|jp|j<|SN)TYPES__name__)subclassclsr2r/r0_inner~sz'Signature.register_type.._innerr/)r=r2r>r/r<r0 register_type|szSignature.register_typecCs:|d}|r.|j|}||k r.|j||dSt||dS)N subtask_typeapp)getr9 from_dictr)r=drBtypZ target_clsr/r/r0rDs   zSignature.from_dictFc  s|||_t|trt|nZz |j} Wntk r@|} YnX||_tj| t|pXd|p`it|phif| ||ddS)Nr/r5argskwargsoptionsr@ immutable) _app isinstancedictsuper__init__r2r,_typetuple) selfr5rHrIrJtyper@rKrBexZ task_name __class__r/r0rPs     zSignature.__init__cOs |||d\}}}|j||S)z0Call the task directly (in the current process).N)_mergerT)rS partial_argspartial_kwargsrHrI_r/r/r0__call__szSignature.__call__cOs |||S)z5Shortcut to :meth:`apply_async` using star arguments. apply_async)rSrYrZr/r/r0delayszSignature.delaycKsP|r|nd}|r|ni}dd|D}||||\}}}|jj||f|S)zCall task locally. Same as :meth:`apply_async` but executed the task inline instead of sending a task message. r/cSsi|]\}}|dk r||qSr8r/.0kvr/r/r0 sz#Signature.apply..)itemsrXrTapply)rSrHrIrJr/r/r0rfs   zSignature.applycKs|r|nd}|r|ni}dd|D}z |j}Wntk rJYdSX|sX|sX|rn||||\}}}n|j|j|j}}}|||f|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. r/cSsi|]\}}|dk r||qSr8r/r`r/r/r0rdsz)Signature.apply_async..N)re _apply_async IndexErrorrXrHrIrJ)rSrHrI route_namerJ_applyr/r/r0r^s    zSignature.apply_asynccs|r|nd}|r|ni}|dk rBtjffdd|D}nj}jr`|s`jj|fS|rvt|tjnj|rtjf|nj|fS)Nr/cs*i|]"\}}|jks |jkr||qSr/)_IMMUTABLE_OPTIONSrJr`rSr/r0rds  z$Signature._merge..)rNrJrerKrHrIrR)rSrHrIrJforce new_optionsr/rlr0rXs    zSignature._mergecKs|r|nd}|r|ni}|s$|s$|r:||||\}}}n|j|j|j}}}tj|jt||t||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. r/rGrA) rXrHrIrJrrDr5rRrr@rKrLrQ)rSrHrIoptsr%r/r/r0clones   zSignature.clonec Cs|j}z |d}Wn&tk r8|p*t}|d<YnX|rF||d<|rR||d<d|krf|jj|d<|rzd|krz||d<|r||d<|dk 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_idZreply_tor7r$N group_index)rJKeyErrorr rBZ thread_oid AsyncResult) rS_idr7r$rrrsrtrotidr/r/r0freeze s$    zSignature.freezecCs6|}|dk r||_|dk r$||_|dk r2||_|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)rprHrIrJ)rSrHrIrJr%r/r/r0replace4szSignature.replacecKs"|dk r|||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_immutablerJupdate)rSrKrJr/r/r0setCs  z Signature.setcCs ||_dSr8rK)rSrKr/r/r0r{OszSignature.set_immutablecCs,|j|g}t|ts(|g}|j|<|Sr8)rJ setdefaultrMr)rSkeyrer/r/r0_with_list_optionRs zSignature._with_list_optioncCs ||}||kr|||Sr8)rappendrSrvaluerer/r/r0append_to_list_optionXs  zSignature.append_to_list_optioncCs||}|t|dSr8)rextendrrr/r/r0extend_list_option^s zSignature.extend_list_optioncCs |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`. linkrrScallbackr/r/r0rbszSignature.linkcCs |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`. link_errorrrSerrbackr/r/r0rkszSignature.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`. rrr/r/r0on_errorts zSignature.on_errorc Cs6ttjt|ggddt|jdp*gDS)zhReturn a recursive list of dependencies. "unchain" if you will, but with links intact. css|]}|VqdSr8) flatten_links)rarr/r/r0 sz*Signature.flatten_links..r)r- itertoolsr from_iterablerrJrCrlr/r/r0rs  zSignature.flatten_linkscCsdt|tr$tt|f||jdSt|trFt|}t|||jdSt|tr`t|||jdSt SNrA) rM_chainr unchain_tasksrLr#r1rBrNotImplementedrSotherr/r/r0__or__s   zSignature.__or__cCs ||Sr8)rrr/r/r0__ior__szSignature.__ior__c Cs~|j}|j}|jdpt}|dL}|j||}|jj |d|j fd|i||j d| |W5QRSQRXdS)Nrqr5) connection) rTrBrJrCr producer_or_acquirebackendZ on_task_callcontrolelectionrprrv)rSrTrBrxproducerpropsr/r/r0rs  zSignature.electioncOs(|j||idd\}}}t|d||S)NT)rmr5)rXr )rSrHrIr[r/r/r0r szSignature.reprcallcCs||t|<t|Sr8)idrN)rSmemor/r/r0 __deepcopy__s zSignature.__deepcopy__cCs |Sr8)r^rCrlr/r/r0 __invert__szSignature.__invert__cCstt|ffSr8)r%rNrlr/r/r0 __reduce__szSignature.__reduce__cCst|Sr8)rNrlr/r/r0__json__szSignature.__json__cCs|Sr8)r rlr/r/r0__repr__szSignature.__repr__c#s4tD]$\}}t|tr$|n||fVq dSr8)rOrerMbytesdecode)rSrbrcrVr/r0reszSignature.itemscCs|jSr8r4rlr/r/r0r2szSignature.namecCs|jp|jj|dSNr5)rQrBr)rlr/r/r0rTszSignature.typecCs |jptSr8)rLrrlr/r/r0rBsz Signature.appcCs,z |jjWStk r&|jjYSXdSr8)rTrvrurBrlr/r/r0rvs zSignature.AsyncResultcCs6z |jjWStk r0t|jj|dYSXdSr)rTr^ru_partialrBZ send_taskrlr/r/r0rgs zSignature._apply_asynczoptions.task_idz Task UUIDzoptions.parent_idzTask parent UUID.zoptions.root_idzTask root UUID.r5z Name of task.rHzPositional arguments to task.rIzKeyword arguments to task.rJzTask execution options.r@zType of signaturerKz+Flag set if no longer accepts new arguments)N)N)NNNNNNFN)NN)NNN)NNNF)NN)NNNNNN)NNN)N);r: __module__ __qualname____doc__r9rLrQrk classmethodr?rDrPr\r_rfr^rXrprry_freezerzr}r{rrrrrrrrrrr rrrrrrepropertyr2r rTrBrvrgrrrsrrr5rHrIrJr@rK __classcell__r/r/rVr0r<s8    !   %                     rcCsJ|rtddi|Sd|kr(td|i|S|dk rFtd|d|i|SdS)Nr r)rJr)use_linkr/r/r0_prepare_chain_from_optionss)rr )r2c seZdZeddZedddZfddZdd Zd d Z fd d Z ddZ d ddZ d!ddZ d"ddZddddddddejdf ddZd#ddZeddZddZZS)$r kwargs.taskszTasks in chain.NcsX|dd}|r@t|tr.t|}|dd<fdd|D}t|fdi|dS)NrIr)csg|]}t|dqSrAr&rar5rAr/r0 1sz$_chain.from_dict..rBrJ)rMrRr-r)r=rErBr)r/rAr0rD+s   z_chain.from_dictcs^t|dkr$t|dr$t|dn|}tjddd|if||dd|_d|_d|_dS)Nr'rz celery.chainr/r)rr ) r(rrrOrPpop _use_linkr@_frozenrSr)rJrVr/r0rP4s$z_chain.__init__cOs|jr|||SdSr8)r)r^)rSrHrIr/r/r0r\=sz_chain.__call__cCst|tr6t|}|}|s"|Stt|||jdSt|tr\tt|||jdSt|tr|j rt|j dtr| }t |j d||jd|j d<|S|j rt|j dt r| }|j dj |B|j d_ |Stt|||jdSnt SdS)NrA)rMr#r1rrrrLrrr)rpr$bodyr)rSrr)sigr/r/r0rAsL    z _chain.__or__cs6ttj||}fdd|jdD|jd<|S)Ncsg|]}|jddqS)T)rBrp)rL)rarrSZ to_signaturer/r0rfsz _chain.clone..r))r&rOrprIrSrHrIr%rVrr0rpcs   z _chain.clonecCs<dd|jD}|jdgD]}|D]}||q&q|S)NcSsg|] }|qSr/rpratr/r/r0rosz(_chain.unchain_tasks..r)r)rJrCr)rSr)rr5r/r/r0rls z_chain.unchain_tasksc Ks||r|nd}|r|ng}|j}|jjrPt|j||f|W5QRSQRX|j||fd|i|rrt|jf|n|jS)Nr/rB)rBconftask_always_eagerrrfrunrNrJ)rSrHrIrJrBr/r/r0r^us  "z_chain.apply_asyncc Ks|r|nd}|r|ng}| p |j} |j}|dkr@| jjdkr@d}|r\|js\t|t|jn|j}|j|||j| | || |||| d \}}|r|r|d d|| }t |||}|j f|}|s|S|dSdS)Nr/r'T)rtrr) rBrr task_protocolrKrRrH prepare_stepsr)rrrr^)rSrHrIr7r$rqrr publisherrrrrsrBrtrJrr)Zresults_from_prepareZ first_taskZresult_from_applyr/r/r0rsD      z _chain.runc Cs<|j|j|j|j||d|j|||d|d \}}|_|dS)NF)rprtr)rrHrIr)rBr) rSrwr7r$rrrsrtr[resultsr/r/r0rysz _chain.freezeTc Cs|p|j}|j}|dkr(|jjdkr(d}t|}|j}|j}d}d}gg}}d}|r|}| | }}t|tj s| ||d}t|t rt |}| r|r| ||}q| }n|rt |t |j|_t|tr||jqRt|t rF|rF||zt|||j||d}Wn&tk rDt||||d}YnX|rb|j||| | | d}n |j|d}|d7}|r|r|||r|js||_|rt|D]}||q||||||}}t|trR|j|}|jr|j}q|}qR||fS) Nr'TrrA)rrqrrrB)rrrrB)rrr7r$rtrr)rBrrrrrrrMrCallableSignaturer#r1rprRrHrr)r$rqr,ryrparentrrrrZensure_chords_allowed)rSrHrIr)rrrsrrBZ last_task_idr7Z chord_bodyrprDrtrZstepsZ steps_popZ steps_extendZ prev_taskZprev_resrir5Z is_first_taskZ is_last_taskresrnoder/r/r0rs                   z_chain.prepare_stepsc Ksz|r|nd}|r|ni}d||f}\}}|jD]D}|||j|oL|fft|jf|}||d|_}\}}q0|S)Nr/)NN)r)rprfrCrNrJr) rSrHrIrJlastZfargsZfkwargsr5rr/r/r0rf.s      z _chain.applycCs<|j}|dkr4z|jdj}Wntk r2YnX|p:tSNr)rLr) LookupErrorrrSrBr/r/r0rB8sz _chain.appcCsJ|js$dt|jdt|ddSt|jddddd |jDS) N<@z#xz: empty>rr5z | css|]}t|VqdSr8)reprrr/r/r0rGsz"_chain.__repr__..)r)rTr:rrjoinrlr/r/r0rBs  z_chain.__repr__)N)NN) NNNNNNNNNNNNN)NNNNNN)NN)r:rrrr)rrDrPr\rrprr^rryrrrfrrBrrr/r/rVr0r'sP   "   '  |  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. cs\|sH|rHt|dks t|drHt|dkr4|dn|}ttj|tStj|f||S)Nr'r)r(rroperatoror_r rO__new__)r=r)rIrVr/r0r{s z chain.__new__)r:rrrrrr/r/rVr0r Js0csBeZdZdZeddZed ddZfddZd dd Z Z S) _basemapNr5itcCs |||dd|i|dSNrIrBrJ) _unpack_argsr=rErBr/r/r0rDsz_basemap.from_dictc s,tj|jd|t|dfddi|dS)Nr/r5rrKT)rOrP _task_namer)rSr5rrJrVr/r0rPs  z_basemap.__init__cKsX|r|nd}|r|ni}||j\}}|jjd|t|dfdt|jdi|S)Nr/rrir5)rrIrTr^r-r6rC)rSrHrIror5rr/r/r0r^s   z_basemap.apply_async)N)NN) r:rrrrrrrDrPr^rr/r/rVr0rs    rc@seZdZdZdZddZdS)r!zMap operation for tasks. Note: Tasks executed sequentially in process, this is not a parallel operation like :class:`group`. z celery.mapcCs.||j\}}d|jdtt|ddS)N[z (x) for x in d]rrIr5rrrSr5rr/r/r0rsz xmap.__repr__Nr:rrrrrr/r/r/r0r!sr!c@seZdZdZdZddZdS)r"z.Map operation for tasks, using star arguments.zcelery.starmapcCs.||j\}}d|jdtt|ddS)Nrz(*x) for x in rrrrr/r/r0rszxstarmap.__repr__Nrr/r/r/r0r"sr"csbeZdZdZedddZedddZfdd Zd d Z dd d Z ddZ edddZ Z S)rz)Partition of tasks into chunks of size n.r5rnNcCs t||dd|i|dSr)rrrr/r/r0rDszchunks.from_dictc s,tjdd|t||dfddi|dS)Nz celery.chunksr/)r5rrrKT)rOrPr)rSr5rrrJrVr/r0rPs zchunks.__init__cKs |jf|Sr8r])rSrJr/r/r0r\szchunks.__call__cKs@|r|nd}|r|ni}|j||fdt|jdi|S)Nr/rir5)r#r^r6rIrC)rSrHrIror/r/r0r^s  zchunks.apply_asynccs:j\}}tfddtt||DjdS)Nc3s|]}t|jdVqdSrAN)r"rL)rapartrSr5r/r0rszchunks.group..rA)rrIr#_chunksiterrL)rSrrr/rr0r#s  z chunks.groupcCs|||||dSrr/)r=r5rrrBr/r/r0 apply_chunksszchunks.apply_chunks)N)NN)N)r:rrrrrrrDrPr\r^r#rrr/r/rVr0rs    rcsxt|trt|d}t|ttfr,|j}nHt|tjr@|g}n4t|trbt fdd|D}nfdd|D}|S)NrAc3s|]}t|dVqdSrr%rrAr/r0rsz_maybe_group..csg|]}t|dqSrrrrAr/r0rsz _maybe_group..) rMrNr%r#rr)rrrr)r)rBr/rAr0 _maybe_groups    rcseZdZdZeddZed1ddZfddZd d Z d d Z d2ddZ d3ddZ d4ddZ ddZddZddZejejeefddZd5ddZdd Zd6d!d"Zd7d#d$ZeZd%d&Zd'd(Zd)d*Zd+d,Zd-d.Z e!d/d0Z"Z#S)8r#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.NcsJ|dd}t|fdd|D|dd<}t|fdi|dS)NrIr)c3s|]}t|dVqdSrrrrAr/r0rsz"group.from_dict..rBrJ)rTr#)r=rErBZ orig_tasksZ rebuilt_tasksr/rAr0rDs  zgroup.from_dictcsnt|dkrL|d}t|tr$|j}t|tjr:|g}t|tsLt|}t j ddd|if|d|_ dS)Nr'rz celery.groupr/r)r#) r(rMr#r)rrrprrrOrPr@rrVr/r0rPs     zgroup.__init__cOs|j|f|Sr8r])rSrYrJr/r/r0r\*szgroup.__call__cCst|||jdS)N)rrB)r$rLrr/r/r0r-sz group.__or__?cCs0t|||dd}|jD]}|jt|dq|S)NT)Z repeatlast) countdown)r r)r}next)rSstartstopsteprr5r/r/r0skew1s z group.skewTcKs|r|nd}|dk rtd|dk r,td|j}|jjrJ|j||f|S|jsX|S||\}} } ||jg| | |} t } t |j | ||| f||d|} |jj | | | d}| t|dkrt|dt r|d}|j}|r|r|||S)Nr/z%Cannot add link to group: use a chordz5Cannot add link to group: do that on individual tasksrHrI)Z ready_barrierr'r)r*rBrrrfr)ry _freeze_gid _preparedr r- _apply_tasksrfinalizer(rMZcurrent_worker_taskZ add_trail)rSrHrI add_to_parentrrrrJrBr7rrr)prresultZ parent_taskr/r/r0r^7s:  zgroup.apply_asyncc snrndrni|j}|js,|S|\}}||jg|||}||fdd|DS)Nr/cs(g|] \}}}|jfdqS)r)rf)rarr[rHrIrJr/r0rcszgroup.apply..)rBr)ryrrr)rSrHrIrJrBr7rrr)r/rr0rf[s  z group.applycCs|jD]}||qdSr8r)r{rSrKr5r/r/r0r{gs zgroup.set_immutablecCs |jdd}|jd|S)NTr~r)rpr}r)rrSrr/r/r0rksz group.linkcstfdd|jDS)Nc3s|]}|VqdSr8r)raZ child_taskrr/r0r~sz#group.link_error..)rRr)r r/r r0rvszgroup.link_errorc cs|D]} || |r| } n || |d} || trR| | j||||} | EdHq|rp| jsp| || | j| _| | j||d|fVqdS)NrA)r7rr)rpr#rr)rKrHry) rSr)rYr7rrrBrrDrMrRr5Zunrollr/r/r0rs       zgroup._preparedc  ks|p|j}||}d} tt|D]\} \} } | \}}}|dk rH|n |jd}| t|7} |dk r| dkr|j || |j f|d|||d| |r|j s|j s|j d7_ |j|dd|Vq&W5QRXdS)Nrr$F)rrr$rHrIr'T)Zweak)rBr enumeraterrJrC_chord_descendrZset_chord_sizer^Z cancelledreadyr.Zthen)rSr)rrBrrr$rHrIrJZ chord_sizeZ task_indexZ current_taskZ next_taskrrr7Z chord_objr/r/r0rs,     zgroup._apply_taskscCs4t|jf|}|dt|d<}|||dfS)Nrqr7rr)rNrJrr rC)rSrJr7r/r/r0rs zgroup._freeze_gidc Cs|j}z |d}Wn&tk r8|p*t}|d<YnX|rF||d<|rR||d<|dk rb||d<|d|}|d|}t|jtrt| |j\} } t | | ||||} t ddt | | D|_n>g} t || ||||} t|jtr| |jdd<n| |_|| fS) Nrqr7r$rtrrrscss|]}|dVqdS)rNr/)raxr/r/r0rsz,group._freeze_group_tasks..)rJrur rrMr)rrtee _unroll_tasksr _freeze_taskszipr-_freeze_unrollr) rSrwr7r$rrrsrtrogidZtasks1Ztasks2r new_tasksr/r/r0_freeze_group_taskss:    zgroup._freeze_group_tasksc Cs|jj|j||||||dS)N)rwr7r$rrrsrt)rBrr)rSrwr7r$rrrsrtr/r/r0rys z group.freezec#s&fddt|DEdHdS)Nc3s&|]\}}|j|dVqdS)r7r$rrrsrtN)ry)rartr5r$r7rsrrr/r0rsz&group._freeze_tasks..)r )rSr)r7r$rrrsr/rr0rszgroup._freeze_tasksc#sfdd|DEdHdS)Nc3s |]}t|jdVqdSr)r&rLrprrlr/r0rsz&group._unroll_tasks..r/rSr)r/rlr0rszgroup._unroll_tasksc csnt|j}d}|rjt||jd}t|tr@||jq| ||j |||||dV|d7}qdS)NrrArr') rr)r&popleftrLrprMr# extendleftrry) rSrr7r$rrrsstackrtr5r/r/r0rs   zgroup._freeze_unrollcCs(|jr$t|jddd|jdSdS)Nrr5zgroup()zgroup())r)rrlr/r/r0rs   zgroup.__repr__cCs t|jSr8)r(r)rlr/r/r0__len__ sz group.__len__cCsD|j}|dkr4z|jdj}Wntk r2YnX|dk r@|StSr)rLr)rBrrrr/r/r0rB#sz group.app)N)rNr)NNTNNN)NN)NNNNNNN)NNNNNN)NNNNNN)$r:rrrrr)rrDrPr\rrr^rfr{rrrrrrMrRrrrrryrrrrrrrrBrr/r/rVr0r#s^    $     * ) r#r$cseZdZdZed/ddZed0ddZd1fdd Zd2d d Z fd d Z d3ddZ d4ddZ d5ddZ eddZddZd6ddZfddZdd Zd!d"Zd#d$Zd%d&Zed'd(Zd7d)d*Zed+d,Zed-d.ZZS)8r 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 NcCs2|}|jf|d\}|d<||d|i|S)NrIrB)copyr)r=rErBrJrHr/r/r0rDJsz_chord.from_dictcKs ||f|fSr8r/)headerrrIr/r/r0rPsz_chord._unpack_args celery.chordc sZ|r|nd}|r|ndii}tj|||t||t||ddfd|i|d|_dS)Nr/rIrA)r!rrBr$)rOrPrr&r@)rSr!rr5rHrIrBrJrVr/r0rPVs    z_chord.__init__cKs|jd|rd|inif|S)Nr/rr])rSrrJr/r/r0r\`sz_chord.__call__cs@t|ttfs0t|tr0|}|j|B|_|St|SdSr8)rMr#rrrprrOr)rSrrrVr/r0rcs z _chord.__or__c Cst|jtst|j|jd|_|jj|||jd}|jj|||||d}|} t} | r| j| krhtd| | j| j dkr|| _ q| j } qR|jj|_|S)NrA)rsrrr$)rrr$r7rtzRecursive result parents) rMr)r#rBryrr}r RuntimeErroraddr) rSrwr7r$rrrsrt header_resultZ body_resultrseenr/r/r0ryms4     z _chord.freezec  Ks$|r|nd}|r|ni}|r4|js4t|t|jn|j}|ddpN|jd} t|jdf|}| jf| } || } t|j t r|j n t |j | d} | j j rt (|j||f| |d| W5QRSQRX| rt|jf| n|j} | dd}|dkr |}|j| | |fd|i| S)Nr/rrIrA)rrqrq)rKrRrHrrIrNrp_get_apprMr)r#rrrrfrJr)rSrHrIrqrrrZrouterZ result_clsrJrrBr)Zmerged_optionsZoption_task_idr/r/r0r^s8        z_chord.apply_asyncTcKsn|r|nd}|r|ni}|dkr&|jn|}t|jtr@|jnt|j|jd}|j|||j|dfdS)Nr/rA) propagate)rH)rrMr)r#rprBrfrC)rSrHrIr(rrJr)r/r/r0rfs  z _chord.applycst|tst|trt|}t|trNt|jd|j}tfdd|DSt|tr|jdddD]} |}|dkrh|SqhdSt|t r |j St|trdSt |S)Nr)c3s|]}|VqdSr8r rr=r/r0rsz"_chord._descend..rrr') rMrrNrDr#r3r)sumrr r$rr()r=Zsig_objZsubtasksZ child_sigZ child_sizer/r*r0r s         z_chord._descendcs&tjdj}tfdd|DS)Nr)c3s|]}|VqdSr8r)rrlr/r0rsz)_chord.__length_hint__..)r3r)r+rr/rlr0r+sz_chord.__length_hint__r'Fc Ks|p ||}|jdpt} |jd} | r>t|jf| n|j} | r`| dd|j| |j| | d} | dd| dd| dd|j| || d}|j r|j j |||||d||d| i| }n| g|j j|}|| _| S)Nrqrrrr r$)r7r$rr)intervalr max_retries)r'rJrCr rNrr|ryrr)rZ apply_chordr_rBrr)rSr!rrYrBr,rr-eagerrqrJr7rrZbodyresZheader_result_argsr%r/r/r0rs2        z _chord.runc sHtj||}zt|jddd|jd<Wnttfk rBYnX|S)NrTr)rOrpr&rIr,rurrVr/r0rpsz _chord.clonecCs|j||Sr8)rrrr/r/r0rs z _chord.linkcCs|j||Sr8)rrrr/r/r0rs z_chord.link_errorcCs|jD]}||qdSr8rrr/r/r0r{ s z_chord.set_immutablec Cs|jrtt|jtrVt|jjddd|jjd|jt|jjdd|jdSdt|jd|j|jSd|jdS) Nrr5z %({} | {!r})r'rA%z) rrMrrr)formatr r rLrlr/r/r0rs  z_chord.__repr__cCs ||jSr8)r'rrlr/r/r0rBsz _chord.appcCsj|j}|dkrZz |jj}Wntk r4|j}YnX|rD|dj}|dkrZ|dk rZ|j}|dk rf|StSr)rLr)r,r)rSrrBr)r/r/r0r' s   z_chord._get_appz kwargs.headerzTasks in chord header.z kwargs.bodyzBody task of chord.)N)NN)Nr"NNN)N)NNNNNN)NNNNNNNN)NNTN)NNr'NFN)N)r:rrrrrD staticmethodrrPr\rryr^rfr r+rrprrr{rr rBr'rr)rrr/r/rVr0r .s^         (   r cOsF|d}t|tr6t|tjr(|Stj||dSt|f||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. rBrA)rCrMrNrrrprrD)ZvariesrHrIrBr/r/r0r%6s   r%FcCsF|dk rBt|tjr"|r4|}nt|tr4t|}|dk rB||_|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)rMrrrprNr%rL)rErBrpr/r/r0r&Js   r&)NF)Grrr collectionsrcollections.abcrr r functoolsrrrrtypesrZkombu.utils.functionalr r Zkombu.utils.objectsr Zkombu.utils.uuidr Zviner Z celery._staterZ celery.resultrrZ celery.utilsrZcelery.utils.collectionsrZcelery.utils.functionalrrrrrrrrrZcelery.utils.objectsrZcelery.utils.textrr__all__r1r6rregisterrNrrr?rr rr!r"rr#r r$r%Zsubtaskr&Z maybe_subtaskr/r/r/r0sr                 49 $< &?