HEX

File: //usr/local/lib/python3.9/site-packages/celery/__pycache__/canvas.cpython-39.pyc
a

X>h�z�@s~dZddlZddlZddlZddlmZmZddlmZddl	m
Z
ddlmZddl
mZddl
mZdd	lmZdd
lmZddlmZmZddlmZdd
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/m0Z0ddl1m2Z2ddl3m4Z4m5Z5dZ6dd�Z7dd�Z8dd�Z9dCd!d"�Z:Gd#d$�d$ed%�Z;e%j<j=Gd&d'�d'e>��Z?d(d)�Z@e?jAd*d+�Gd,d-�d-e?��ZBGd.d*�d*eB�ZCGd/d0�d0e?�ZDe?�A�Gd1d2�d2eD��ZEe?�A�Gd3d4�d4eD��ZFe?�A�Gd5d6�d6e?��Z*d7d8�ZGe?�A�Gd9d:�d:e?��ZHe?jAd;d+�Gd<d=�d=e?��ZIeIZJd>d?�ZKeKZLdDdAdB�ZMeMZNdS)EzmComposing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
�N)�ABCMeta�abstractmethod)�deque)�MutableSequence)�deepcopy)�partial)�reduce)�
itemgetter)�
GeneratorType)�fxrange�reprcall)�cached_property)�uuid)�barrier)�current_app)�CPendingDeprecationWarning)�GroupResult�allow_join_result)�abstract��ChainMap)�_regen)�chunks)�is_list�
maybe_list�regen�seq_concat_item�seq_concat_seq)�getitem_property)�remove_repeating_from_task�truncate)	�	Signature�chain�xmap�xstarmapr�group�chord�	signature�maybe_signaturecCs�zt|j�}Wn^tylz|j��}WnttfyH|YYS0|dkr`t|j�dn|YSYn0|dkr�|jdS|SdS)z�Unroll group with only one member.
    This allows treating a group of a single task as if it
    was a single task without pre-knowledge.�rN)�len�tasks�	TypeError�__length_hint__�AttributeError�list)r%�size�r1�7/usr/local/lib/python3.9/site-packages/celery/canvas.py�maybe_unroll_group*s$r3cCst|d|�S)N�name)�getattr��taskr1r1r2�task_name_from<sr8cKs|j||fi|��|S)z�When stamping a sequence of tasks created by a generator,
    we use this function to stamp each task in the generator
    without exhausting it.)�stamp)r7�visitor�
append_stamps�headersr1r1r2�_stamp_regen_task@sr=TcCs�|sdS|��D]�\}}||vrt|t�r>t||||�qt|tttf�r`|rX|gn|||<t||t�r�t||t�r�||�||�q|r||dur�g||<nt||�||<||�	||�q|��D]\}}||vr�|||<q�dS)a�Merge two dictionaries recursively into the first one.

    Example:
    >>> 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�
isinstance�dict�_merge_dictionaries�int�float�strr/�extend�append)Zd1Zd2�aggregate_duplicates�key�valuer1r1r2rAIs$

rAc@s�eZdZdZed�dd�Zdd�dd�Zed�dd	�Zdd�d
d�Ze	ed�dd
��Z
ed�dd�Zdd�dd�Zed�dd�Z
ed�dd�Zed�dd�ZdS)�StampingVisitorz�Stamping 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.
    )�returncKsiS)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.
         r1��selfr%r<r1r1r2�on_group_start|s	zStampingVisitor.on_group_startNcKsdS)z�Method 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.
         Nr1rLr1r1r2�on_group_end�szStampingVisitor.on_group_endcKsiS)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.
         r1�rMr"r<r1r1r2�on_chain_start�s	zStampingVisitor.on_chain_startcKsdS)z�Method 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.
         Nr1rPr1r1r2�on_chain_end�szStampingVisitor.on_chain_endcKsdS)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.
         Nr1)rM�sigr<r1r1r2�on_signature�szStampingVisitor.on_signaturecKs,t|jt�st|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.
         )r?r+r%rN�rMrS�headerr1r1r2�on_chord_header_start�s	z%StampingVisitor.on_chord_header_startcKs|j|jfi|��dS)u�Method 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)rOr+rUr1r1r2�on_chord_header_end�sz#StampingVisitor.on_chord_header_endcKsiS)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.
        r1rUr1r1r2�
on_chord_body�s	zStampingVisitor.on_chord_bodycKsiS)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.
         r1)rM�callbackrVr1r1r2�on_callback�s	zStampingVisitor.on_callbackcKsiS)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.
         r1)rM�errbackrVr1r1r2�
on_errback�s	zStampingVisitor.on_errback)�__name__�
__module__�__qualname__�__doc__r@rNrOrQrRrrTrWrXrYr[r]r1r1r1r2rJvs		

	rJ)�	metaclasscs�eZdZdZiZdZZddhZededd��Z	edfdd��Z
dg�fd
d�	Zdd
�Zdd�Z
dhdd�Zdidd�Zdjdd�Zdkdd�ZeZdldd�ZeZdmdd�Zdndd�Zdd�Zdod!d"�Zdpd#d$�Zdqd%d&�Zd'd(�Zd)d*�Zd+d,�Zd-d.�Zd/d0�Zd1d2�Z d3d4�Z!d5d6�Z"d7d8�Z#d9d:�Z$d;d<�Z%d=d>�Z&d?d@�Z'dAdB�Z(dCdD�Z)dEdF�Z*�fdGdH�Z+e,dIdJ��Z-e.dKdL��Z/e.dMdN��Z0e.dOdP��Z1e.dQdR��Z2e3dSdT�Z4e3dUdV�Z5e3dWdX�Z6e3dYdZ�Z7e3d[d\�Z8e3d]d^�Z9e3d_d`�Z:e3dadb�Z;e3dcdd�Z<�Z=S)rr!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={}}
    N�group_id�stamped_headerscs��fdd�}|S)z�Register a new type of signature.
        Used as a class decorator, for example:
        >>> @Signature.register_type()
        >>> class mysig(Signature):
        >>>     pass
        cs|�j�p|j<|S�N)�TYPESr^)�subclass��clsr4r1r2�_inner/sz'Signature.register_type.<locals>._innerr1)rir4rjr1rhr2�
register_type'szSignature.register_typecCs:|�d�}|r.|j|}||ur.|j||d�St||d�S)z�Create a new signature from a dict.
        Subclasses can override this method to customize how are
        they created from a dict.
        �subtask_type��app)�getrf�	from_dictr!)ri�drn�typZ
target_clsr1r1r2rp5s

zSignature.from_dictFc	s~||_t|t�rt��|�n\z
|j}
Wnty>|}
Yn0||_t�j|
t|pVd�|p^it|pfifi|	��||d�dS)Nr1�r7�args�kwargs�optionsrl�	immutable)	�_appr?r@�super�__init__r4r.�_type�tuple)rMr7rtrurv�typerlrwrn�exZ	task_name��	__class__r1r2rzBs


�zSignature.__init__cOs$|�||d�\}}}|j|i|��S)z0Call the task directly (in the current process).N)�_merger})rM�partial_args�partial_kwargsrtru�_r1r1r2�__call__ZszSignature.__call__cOs|�||�S)z5Shortcut to :meth:`apply_async` using star arguments.��apply_async)rMr�r�r1r1r2�delay_szSignature.delaycKsT|r|nd}|r|ni}dd�|��D�}|�|||�\}}}|jj||fi|��S)z�Call task locally.

        Same as :meth:`apply_async` but executed the task inline instead
        of sending a task message.
        r1cSsi|]\}}|dur||�qSrer1��.0�k�vr1r1r2�
<dictcomp>l�z#Signature.apply.<locals>.<dictcomp>)r>r�r}�apply)rMrtrurvr1r1r2r�cs
zSignature.applycKs�|r|nd}|r|ni}dd�|��D�}z
|j}WntyHYdS0|sV|sV|rl|�|||�\}}}n|j|j|j}}}|||fi|��S)a�Apply 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:`[email protected]_async` and the :ref:`guide-calling` guide.
        r1cSsi|]\}}|dur||�qSrer1r�r1r1r2r��r�z)Signature.apply_async.<locals>.<dictcomp>N)r>�_apply_async�
IndexErrorr�rtrurv)rMrtru�
route_namerv�_applyr1r1r2r�qs
zSignature.apply_asynccs�|r|nd}|r|ni}|durn�j�d�jvrJ�j�t�j�dg����i�j���fdd�|��D��}n�j}�jr�|s��j�j|fS|r�t	|�t	�j�n�j|r�t
�jfi|��n�j|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)
        r1Nrdcs(i|] \}}|�vs|�jvr||�qSr1)rvr��Zimmutable_optionsrMr1r2r��s�z$Signature._merge.<locals>.<dictcomp>)�_IMMUTABLE_OPTIONSrv�union�setror>rwrtrur|r@)rMrtrurv�force�new_optionsr1r�r2r��s 
�

�zSignature._mergecKs�|r|nd}|r|ni}|s$|s$|r:|�|||�\}}}n|j|j|j}}}tj|jt|�|t|�|j	|j
d�|jd�}|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.
        r1rsrm)
r�rtrurvr!rpr7r|rrlrwrxr{)rMrtru�optsr'r1r1r2�clone�s 	��zSignature.clonec	Cs�|j}z|d}Wn$ty6|p(t�}|d<Yn0|rD||d<|rP||d<d|vrd|jj|d<|rxd|vrx||d<|r�||d<|dur�||d<|�|�S)	aCFinalize 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.

        The arguments are used to override the signature's headers during
        freezing.

        Arguments:
            _id (str): Task id to use if it didn't already have one.
                New UUID is generated if not provided.
            group_id (str): Group id to use if it didn't already have one.
            chord (Signature): Chord body when freezing a chord header.
            root_id (str): Root id to use.
            parent_id (str): Parent id to use.
            group_index (int): Group index to use.

        Returns:
            ~@AsyncResult: promise of future evaluation.
        �task_id�root_id�	parent_idZreply_torcr&N�group_index)rv�KeyErrorrrnZ
thread_oid�AsyncResult)	rM�_idrcr&r�r�r�r��tidr1r1r2�freeze�s$zSignature.freezecCs6|��}|dur||_|dur$||_|dur2||_|S)z�Replace 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�rtrurv)rMrtrurvr'r1r1r2�replaceszSignature.replacecKs"|dur|�|�|j�|�|S)u�Set arbitrary execution options (same as ``.options.update(…)``).

        Returns:
            Signature: This is a chaining method call
                (i.e., it will return ``self``).
        N)�
set_immutablerv�update)rMrwrvr1r1r2r�s
z
Signature.setcCs
||_dSre�rw)rMrwr1r1r2r�&szSignature.set_immutableTcs�����d�vr t�����d<|durj|p.i}d|vrHt|���|d<t�||d�tt�d���d<|r�t��dg��}|�|j�dg��t|��d<�fdd�|j��D�}t�||d�tt�d���d<�S)a�Collect all stamps from visitor, headers and self,
        and return an idempotent dictionary of stamps.

        .. versionadded:: 5.3

        Arguments:
            visitor_headers (Dict): Stamps from a visitor method.
            append_stamps (bool):
                If True, duplicated stamps will be appended to a list.
                If False, duplicated stamps will be replaced by the last stamp.
            self_headers (bool):
                If True, stamps from self.options will be added.
                If False, stamps from self.options will be ignored.
            headers (Dict): Stamps that should be added to headers.

        Returns:
            Dict: Merged stamps.
        rdN)rGcs"i|]\}}|�dvr||�qS)rdr1r��r<r1r2r�]r�z,Signature._stamp_headers.<locals>.<dictcomp>)	�copyr/�keysrAr�ror�rvr>)rM�visitor_headersr;�self_headersr<rdZredacted_optionsr1r�r2�_stamp_headers)s"zSignature._stamp_headerscKsb|j||fi|��|��}d}|dur>|j|fi|��p<i}|j||fi|��}|jfi|��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.
        N)�stamp_linksr�rTr�r�)rMr:r;r<r�r1r1r2r9eszSignature.stampcKs|��}d}t|�}t|j�d��p&gD]`}t||jd�}d}|dur\|j|fi|��pZi}|jf|||d�|��}|j	||fi|��q(t|�}t|j�d��p�gD]`}t||jd�}d}|dur�|j
|fi|��p�i}|jf|||d�|��}|j	||fi|��q�dS)a�Stamp 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.
        F�linkrmN)r�r;r��
link_error)r�rrrvror(rnr[r�r9r])rMr:r;r<Znon_visitor_headersr�r�r�r1r1r2r�{s<����zSignature.stamp_linkscCs,|j�|g�}t|t�s(|g}|j|<|S)a�Gets 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.
        )rv�
setdefaultr?r)rMrHr>r1r1r2�_with_list_option�s
zSignature._with_list_optioncCs |�|�}||vr|�|�|S)zEAppends the given value to the list at the given key in self.options.)r�rF�rMrHrIr>r1r1r2�append_to_list_option�s

zSignature.append_to_list_optioncCs|�|�}|�t|��dS)z�Extends 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)r�rErr�r1r1r2�extend_list_option�s
zSignature.extend_list_optioncCs|�d|�S)z�Add callback task to be applied if this task succeeds.

        Returns:
            Signature: the argument passed, for chaining
                or use with :func:`~functools.reduce`.
        r��r��rMrZr1r1r2r��szSignature.linkcCs|�d|�S)z�Add callback task to be applied on error in task execution.

        Returns:
            Signature: the argument passed, for chaining
                or use with :func:`~functools.reduce`.
        r�r��rMr\r1r1r2r��szSignature.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`.
        )r�r�r1r1r2�on_error�s

zSignature.on_errorcCs6ttj�t�|ggdd�t|j�d��p*gD����S)zhReturn a recursive list of dependencies.

        "unchain" if you will, but with links intact.
        css|]}|��VqdSre)�
flatten_links)r�r�r1r1r2�	<genexpr>�s�z*Signature.flatten_links.<locals>.<genexpr>r�)r/�	itertoolsr"�
from_iterablerrvro�rMr1r1r2r��s��zSignature.flatten_linkscCsdt|t�r$tt|f|���|jd�St|t�rFt|�}t|||jd�St|t�r`t|||jd�St	S)z�Chaining operator.

        Example:
            >>> add.s(2, 2) | add.s(4) | add.s(8)

        Returns:
            chain: Constructs a :class:`~celery.canvas.chain` of the given signatures.
        rm)
r?�_chainr�
unchain_tasksrxr%r3rnr!�NotImplemented�rM�otherr1r1r2�__or__�s	

��

zSignature.__or__cCs
|�|�Sre)r�r�r1r1r2�__ior__
szSignature.__ior__cCs�|j}|j}|j�d�pt�}|�d��N}|j�||�}|jj	|d|j
fd|i|��|jd�|�|�Wd�S1s|0YdS)Nr�r7)�
connection)
r}rnrvror�producer_or_acquire�backendZon_task_call�control�electionr�r�r�)rMr}rnr��producer�propsr1r1r2r�s
�zSignature.electioncOs(|j||idd�\}}}t|d||�S)aAReturn a string representation of the signature.

        Merges the given arguments with the signature's arguments
        only for the purpose of generating the string representation.
        The signature itself is not modified.

        Example:
            >>> add.s(2, 2).reprcall()
            'add(2, 2)'
        T)r�r7)r�r)rMrtrur�r1r1r2rszSignature.reprcallcCs||t|�<t|�Sre)�idr@)rM�memor1r1r2�__deepcopy__-szSignature.__deepcopy__cCs|����Sre)r�ror�r1r1r2�
__invert__1szSignature.__invert__cCstt|�ffSre)r'r@r�r1r1r2�
__reduce__4szSignature.__reduce__cCst|�Sre)r@r�r1r1r2�__json__9szSignature.__json__cCs|��Sre)rr�r1r1r2�__repr__<szSignature.__repr__c#s4t���D]$\}}t|t�r$|��n||fVq
dSre)ryr>r?�bytes�decode)rMr�r�rr1r2r>?szSignature.itemscCs|jSrer6r�r1r1r2r4CszSignature.namecCs|jp|jj|dS�Nr7)r{rnr+r�r1r1r2r}HszSignature.typecCs
|jptSre)rxrr�r1r1r2rnLsz
Signature.appcCs*z
|jjWSty$|jjYS0dSre)r}r�r�rnr�r1r1r2r�Ps
zSignature.AsyncResultcCs4z
|jjWSty.t|jj|d�YS0dSr�)r}r�r��_partialrnZ	send_taskr�r1r1r2r�Ws
zSignature._apply_asynczoptions.task_idz	Task UUIDzoptions.parent_idzTask parent UUID.zoptions.root_idzTask root UUID.r7z
Name of task.rtzPositional arguments to task.ruzKeyword arguments to task.rvzTask execution options.rlzType of signaturerwz+Flag set if no longer accepts new arguments)N)N)NNNNNNFN)NN)NNN)NNNF)NN)NNNNNN)NNN)N)NFT)NF)F)>r^r_r`rarfrxr{r��classmethodrkrprzr�r�r�r�r�r�rr��_freezer�r�r�r�r9r�r�r�r�r�r�r�r�r�r�r�rr�r�r�r�r�r>�propertyr4r
r}rnr�r�rr�r�r�r7rtrurvrlrw�
__classcell__r1r1rr2r!�s�8
�

!
*
�
1


<

1		













�r!cCsJ|rtddi|�Sd|vr(td|i|�S|durFtd|d|i|�SdS)Nr"r)rvr+�use_linkr1r1r2�_prepare_chain_from_optionsjs)�r�r")r4c
s�eZdZedd�Zed"dd��Z�fdd�Zdd	�Zd
d�Z	�fdd
�Z
dd�Zd#dd�Zd$dd�Z
d%dd�Zd&dd�Zddddddddejdf
dd�Zd'dd�Zedd��Zd d!�Z�ZS)(r��kwargs.taskszTasks in chain.NcsX|dd}|r@t|t�r.t|�}|dd<�fdd�|D�}||fd�i|d��S)Nrur+csg|]}t|�d��qS�rm�r(�r�r7rmr1r2�
<listcomp>�r�z$_chain.from_dict.<locals>.<listcomp>rnrv)r?r|r/)rirqrnr+r1rmr2rp�s
z_chain.from_dictcsbt|�dkr$t|d�r$t|d�n|}t�jddd|ifi|��|�dd�|_d|_d|_dS)Nr)rzcelery.chainr1r+r�r")	r*rrryrz�pop�	_use_linkrl�_frozen�rMr+rvrr1r2rz�s$�z_chain.__init__cOs|jr|�||�SdSre)r+r�)rMrtrur1r1r2r��sz_chain.__call__cCs�t|t�r�t|�}t|t�s&|�|�S|��}|s6|St|dt�rj|dj|B|d_t|�||jd�St|�t	||�|j
d�St|t�r�tt
j|��|�St|t��r�|j�r<t|jdt��r<|��}t|jd||j
d�|jd<t|j�dk�r8t|jdt��r8|jdj|jdB|jd_|jdd�|_|S|j�rzt|jdt��rz|��}|jdj|B|jd_|St|�t	|��|�|j
d�SntSdS)N���rmr)���)r?r%r3r�r�r&�bodyr}rnrrxr�r�operator�or_r!r+r�r*r�)rMr�r+rSr1r1r2r��sJ


��
�"��z
_chain.__or__cs:t�t�j|i|��}��fdd�|jdD�|jd<|S)Ncsg|]}�|�jdd��qS)T)rnr�)rx)r�rS�rMZto_signaturer1r2r��s�z _chain.clone.<locals>.<listcomp>r+)r(ryr�ru�rMrtrur'rr�r2r��s�z_chain.clonecCsjdd�|jD�}t|j�d��p"gD]}|d�|�q$t|j�d��pJgD]}|D]}|�|�qTqL|S)a�Return 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.
        cSsg|]}|���qSr1�r��r��tr1r1r2r�r�z(_chain.unchain_tasks.<locals>.<listcomp>r�r�r�)r+rrvror�r�)rMr+rSr7r1r1r2r��sz_chain.unchain_taskscKs�|r|nd}|r|ng}|j}|jjr`t��$|j||fi|��Wd�S1sV0Y|j||fd|i|r�t|jfi|��n|j��S)Nr1rn)rn�conf�task_always_eagerrr��runr@rv)rMrtrurvrnr1r1r2r�	s2�z_chain.apply_asyncc
Ks�|r|nd}|r|ng}|p |j}|j}|dur@|jjdkr@d}|r\|js\t|�t|j�n|j}|j|||j|
|||||||
d�\}}|r�|r�|d�	d|�|�
�}t|||�}|jfi|��}|s�|S|dSdS)z�Executes 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.
        r1Nr)T)r�rr�)
rnr�r��
task_protocolrwr|rt�
prepare_stepsr+r�r�r�r�)rMrtrurcr&r�r�r��	publisherr�r�r�rnr�rvr�r+Zresults_from_prepare�
first_taskZresult_from_applyr1r1r2r�s2
����
z
_chain.runc	Cs<|j|j|j|j||d|j|||d|d�\}}|_|dS)NF)r�r�r)r�rtrur+rnr�)	rMr�rcr&r�r�r�r��resultsr1r1r2r�Es�z
_chain.freezeFcKs�d}|dur"|j|fi|��p i}|j||fi|��}|j|fi|��|jD]}|j||fi|��qN|dur�|j|fi|��dSre)rQr�r�r+r9rR)rMr:r;r<r�r7r1r1r2r9Ps
z_chain.stampTc	Cs:|p|j}|j}|dur(|jjdkr(d}t|�}|j}|j}d}d}gg}}d}|�r,|�}||}}t|tj	�s�|||d�}t|t
�r�t|�}|r�|r�|�||�}q�|��}n|r�t
|�t
|j�|_t|t�r�||j�qRt|t
��rd|�rd|��|��zt|||j||d�}Wn$t�yBt||||d�}Yn0|�r\|d}|d}nd}d}|�r�|j|||	|
|
d	�}n|j|d
�}|d7}|�r�|�r�|�|�|�r�|j�s�||_|�r�t|�D]}|�|��q�|�|�|�|�||}}t|t�rR|j��|}|j�r&|j}�q|}qR||_||fS)a�Prepare 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)Trrm)r�r�r�rn)r�r�rnr�)r�rcr&r��r�)rnr�r�r�rr�rEr?r�CallableSignaturer%r3r�r|rtr�r+r&r�r.r�r��parentrr�rFr�Zensure_chords_allowedr�)rMrtrur+r�r�r�rnZlast_task_idrcZ
chord_bodyr�rpr�r�ZstepsZ	steps_popZsteps_extendZ	prev_taskZprev_resr��ir7Z
is_first_taskZis_last_task�resr\�noder1r1r2r�]s�%





�
�
�






z_chain.prepare_stepsc	Ks�|r|nd}|r|ni}d||f}\}}|jD]L}|�||�j|oL|��ffit|jfi|����}||d|_}\}}q0|S)Nr1)NN)r+r�r�ror@rvr�)	rMrtrurv�lastZfargsZfkwargsr7r�r1r1r2r�s
��z_chain.applycCs:|j}|dur2z|jdj}Wnty0Yn0|p8tS�Nr)rxr+�LookupErrorr�rMrnr1r1r2rn
sz
_chain.appcCsJ|js$dt|�j�dt|�d�d�St|jddd�dd	�|jD���S)
N�<�@z#xz: empty>rr7z | css|]}t|�VqdSre)�reprr�r1r1r2r�r�z"_chain.__repr__.<locals>.<genexpr>)r+r}r^r�r�joinr�r1r1r2r�s�z_chain.__repr__)N)NN)
NNNNNNNNNNNNN)NNNNNN)NF)NN)r^r_r`rr+r�rprzr�r�r�r�r�r�r�r9r!r�r�r�rnr�r�r1r1rr2r��s8
	0	
�
0�

�
$


	r�cs eZdZdZ�fdd�Z�ZS)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�|s�|r�t|�dks t|d�r�t|�dkr4|dn|}ttj|t��}|tkr|t|t�r|t||�s|t�j	||j
fi|��S|St�j	|g|�Ri|��S)Nr)r)r*rrr�r�r�r"r?ry�__new__r+)rir+ruZnew_instancerr1r2rMsz
chain.__new__)r^r_r`rarr�r1r1rr2r"s0csBeZdZdZedd�Zed
dd��Z�fdd�Zddd	�Z	�Z
S)�_basemapNr7�itcCs ||�|d�d|i|d��S�Nrurnrv��_unpack_args�rirqrnr1r1r2rp_sz_basemap.from_dictcs,t�j|jd|t|�d�fddi|��dS)Nr1�r7r
rwT)ryrz�
_task_namer)rMr7r
rvrr1r2rzcs���z_basemap.__init__cKsX|r|nd}|r|ni}|�|j�\}}|jjd|t|�d�fdt|j�d��i|��S)Nr1rr�r7)r
rur}r�r/r8ro)rMrtrur�r7r
r1r1r2r�hs���z_basemap.apply_async)N)NN)r^r_r`rr	r
r�rprzr�r�r1r1rr2r	[s
r	c@seZdZdZdZdd�ZdS)r#z�Map operation for tasks.

    Note:
        Tasks executed sequentially in process, this is not a
        parallel operation like :class:`group`.
    z
celery.mapcCs.|�|j�\}}d|j�dtt|�d��d�S)N�[z
(x) for x in �d�]�r
rur7r r�rMr7r
r1r1r2r�~sz
xmap.__repr__N�r^r_r`rarr�r1r1r1r2r#ssr#c@seZdZdZdZdd�ZdS)r$z.Map operation for tasks, using star arguments.zcelery.starmapcCs.|�|j�\}}d|j�dtt|�d��d�S)Nrz(*x) for x in rrrrr1r1r2r��szxstarmap.__repr__Nrr1r1r1r2r$�sr$csbeZdZdZeddd�Zeddd��Z�fdd	�Zd
d�Z	ddd
�Z
dd�Zeddd��Z�Z
S)rz)Partition of tasks into chunks of size n.r7r
�nNcCs ||�|d�d|i|d��Srrrr1r1r2rp�szchunks.from_dictcs,t�jdd|t|�|d�fddi|��dS)Nz
celery.chunksr1)r7r
rrwT)ryrzr)rMr7r
rrvrr1r2rz�s
���zchunks.__init__cKs|jfi|��Srer�)rMrvr1r1r2r��szchunks.__call__cKs@|r|nd}|r|ni}|��j||fdt|j�d��i|��S)Nr1r�r7)r%r�r8ruro)rMrtrur�r1r1r2r��s���zchunks.apply_asynccs:���j�\�}}t��fdd�tt|�|�D��jd�S)Nc3s|]}t�|�jd�VqdS�rmN)r$rx)r��part�rMr7r1r2r��s�zchunks.group.<locals>.<genexpr>rm)r
rur%�_chunks�iterrx)rMr
rr1rr2r%�s��zchunks.groupcCs|||||d��S)Nrmr1)rir7r
rrnr1r1r2�apply_chunks�szchunks.apply_chunks)N)NN)N)r^r_r`rar	r
r�rprzr�r�r%rr�r1r1rr2r�s
rcsxt|t�rt|�d�}t|ttf�r,|j}nHt|tj�r@|g}n4t|t�rbt	�fdd�|D��}n�fdd�|D�}|S)Nrmc3s|]}t|�d�VqdSr�r'r�rmr1r2r��r�z_maybe_group.<locals>.<genexpr>csg|]}t|�d��qSr�rr�rmr1r2r��r�z _maybe_group.<locals>.<listcomp>)
r?r@r'r%r�r+rr�r
r)r+rnr1rmr2�_maybe_group�s

rcs�eZdZdZedd�Zed4dd��Z�fdd�Zd	d
�Z	dd�Z
d5dd�Zd6dd�Zd7dd�Z
dd�Zd8dd�Zdd�Zdd�Zejejeefdd�Zd9d d!�Zd"d#�Zd:d$d%�Zd;d&d'�ZeZd(d)�Zd*d+�Zd,d-�Zd.d/�Z d0d1�Z!e"d2d3��Z#�Z$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.NcsJ|dd}t|��fdd�|D��|dd<}||fd�i|d��S)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).
        rur+c3s|]}t|�d�VqdSrr�r�rmr1r2r�sz"group.from_dict.<locals>.<genexpr>rnrv)r})rirqrnZ
orig_tasksZ
rebuilt_tasksr1rmr2rp�s
2�zgroup.from_dictcsrt|�dkrL|d}t|t�r$|j}t|tj�r:|��g}t|t�sLt|�}t	�j
ddd|ifi|��d|_dS)Nr)rzcelery.groupr1r+r%)r*r?r%r+rr�r�rrryrzrlr�rr1r2rzs


zgroup.__init__cOs|j|fi|��Srer�)rMr�rvr1r1r2r�-szgroup.__call__cCst|||jd�S)N)r�rn)r&rxr�r1r1r2r�0szgroup.__or__��?cCs0t|||dd�}|jD]}|jt|�d�q|S)NT)Z
repeatlast)�	countdown)rr+r��next)rM�start�stop�stepr
r7r1r1r2�skew4s
z
group.skewTcKs|r|nd}|durtd��|dur,td��|j}|jjrN|j||fi|��S|js\|��S|�|�\}}	}
|�|jg|	|
|�}t	�}t
|j||||f||d�|���}
|jj|	|
|d�}|�
�t|�dkr�t|dt�r�|d}|j}|r�|r�|�|�|S)Nr1z%Cannot add link to group: use a chordz5Cannot add link to group: do that on individual tasks�rtru)Z
ready_barrierr)r)r,rnr�r�r�r+r��_freeze_gid�	_preparedrr/�_apply_tasksr�finalizer*r?Zcurrent_worker_taskZ	add_trail)rMrtru�
add_to_parentr�r�r�rvrnrcr�r+�pr��resultZparent_taskr1r1r2r�;s8���
zgroup.apply_asynccsn�r�nd��r�ni�|j}|js,|��S|���\�}}|�|jg|||�}|�|���fdd�|D��S)Nr1cs(g|] \}}}|jf��d�����qS)r')r�)r�rSr��rtrurvr1r2r�gszgroup.apply.<locals>.<listcomp>)rnr+r�r(r)r)rMrtrurvrnrcr�r+r1r/r2r�_s�zgroup.applycCs|jD]}|�|�qdSre�r+r��rMrwr7r1r1r2r�ks
zgroup.set_immutableFcKs�d}|dur"|j|fi|��p i}|j||fi|��}|j||fi|��t|jt�rv|j�ttf||d�|���n^g}|jD]0}t	||j
d�}|j||fi|��|�|�q�t|jt
�r�||jdd�<n||_|dur�|j|fi|��dS)N�r:r;rm)rNr�r�r?r+r�mapr�r=r(rnr9rFrrO)rMr:r;r<r��	new_tasksr7r1r1r2r9os" 
zgroup.stampcCs |��jdd�}|jd�|�S)NTr�r)r�r�r+r��rMrSr1r1r2r��sz
group.linkcs t���t�fdd�|jD��S)Nc3s |]}|��jdd��VqdS)Tr�N)r�r�)r�Z
child_task�rSr1r2r��r�z#group.link_error.<locals>.<genexpr>)r(r|r+r5r1r6r2r��szgroup.link_errorc

cs�t|�D]�\}
}|||�r$|��}n|||d�}||t�rZ|�|j||||�}|EdHq|rx|jsx|	|�|	|j�|_||j|||
d�|fVqdS)a�Recursively 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)``.
        rmN)rcr�r�)�	enumerater�r%r)r+rwrtr�)
rMr+r�rcr�rnr�rpr?r|�indexr7Zunrollr1r1r2r)�s


�
zgroup._preparedc	
ks|p|j}|�|���}d}
t�|�\}}t|d�t|d�}t|�D]�\}
}|\}}}|durb|n
|j�d�}|
t�	|�7}
|dur�|dur�|j
�||
�|jf|d|||d�|	��|r�|j
s�|js�|jd7_|j|dd�t|d�}|VqDWd�n1�s
0YdS)	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)r�r,r&rtrur)T)Zweak)rnr�r��teer"r7rvro�_chord�_descendr�Zset_chord_sizer��	cancelled�readyr0Zthen)rMr+r�rnr-r,r&rtrurvZ
chord_sizeZ
tasks_shifted�	next_taskZ
task_index�current_taskrSr�rcZ	chord_objr1r1r2r*�s,




��	
zgroup._apply_taskscsFi�j��fdd�|��D��}|�dt��|d<}|||�d�fS)z:Freeze the group id by the existing task_id or a new UUID.cs*i|]"\}}|�jvs |�jvr||�qSr1)r�rvr�r�r1r2r�s�z%group._freeze_gid.<locals>.<dictcomp>r�rcr�)rvr>r�rro)rMrvrcr1r�r2r(s��
zgroup._freeze_gidc
Cs|j}z|d}Wn$ty6|p(t�}|d<Yn0|rD||d<|rP||d<|dur`||d<|�d|�}|�d|�}t|jt�r�t�|�	|j��\}	}
t
|�|	||||��}t
|
�|_n<g}t|�
|||||��}t|jt�r�||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.
        r�rcr&Nr�r�r�)rvr�rr�r?r+rr�r9�
_unroll_tasksr�
_freeze_tasksr/�_freeze_unrollr)
rMr�rcr&r�r�r�r��gidZtasks1Ztasks2r�r4r1r1r2�_freeze_group_taskss2

�zgroup._freeze_group_tasksc	Cs|jj|j||||||d��S)N)r�rcr&r�r�r�)rnrrD)rMr�rcr&r�r�r�r1r1r2r�Ms
�zgroup.freezec#s&����fdd�t|�D�EdHdS)zKCreates a generator for the AsyncResult of each task in the tasks argument.c3s&|]\}}|j����|d�VqdS)�rcr&r�r�r�N)r�)r�r�r7�r&rcr�r�r1r2r�Xs��z&group._freeze_tasks.<locals>.<genexpr>N)r7)rMr+rcr&r�r�r1rFr2rAVs�zgroup._freeze_tasksc#s�fdd�|D�EdHdS)z?Creates a generator for the cloned tasks of the tasks argument.c3s |]}t|�jd���VqdSr)r(rxr�r�r�r1r2r�br�z&group._unroll_tasks.<locals>.<genexpr>Nr1�rMr+r1r�r2r@_szgroup._unroll_tasksc	csnt|j�}d}|rjt|��|jd���}t|t�r@|�|j�q|�	|�|j
|||||d�V|d7}qdS)a�Generator 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.
        rrmrEr)N)rr+r(�popleftrxr�r?r%�
extendleftrFr�)	rMr4rcr&r�r��stackr�r7r1r1r2rBds


�zgroup._freeze_unrollcCs(|jr$t|jddd|j�d��SdS)Nrr7zgroup(�)zgroup(<empty>))r+rr�r1r1r2r��s�zgroup.__repr__cCs
t|j�Sre)r*r+r�r1r1r2�__len__�sz
group.__len__cCsB|j}|dur2z|jdj}Wnty0Yn0|dur>|StSr)rxr+rnrrrr1r1r2rn�sz	group.app)N)r Nr )NNTNNN)NN)NF)NNNNNNN)NNNNNN)NNNNNN)%r^r_r`rarr+r�rprzr�r�r&r�r�r�r9r�r�rr�r!r?r|r)r*r(rDr�r�rAr@rBr�rLr�rnr�r1r1rr2r%�sN
7
�
$


�
1�
@�
3�
	$r%r&cs�eZdZdZed1dd��Zed2dd��Zd3�fdd	�	Zd4d
d�Z	�fdd
�Z
d5dd�Zd6dd�Zd7dd�Z
d8dd�Zedd��Zdd�Zd9dd�Z�fdd �Zd!d"�Zd#d$�Zd%d&�Zd'd(�Zed)d*��Zd:d+d,�Zed-d.�Zed/d0�Z�ZS);r: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
    NcCs6|��}|jfi|d��\}|d<||d|i|��S)a�Create 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).
        rurn)r�r
)rirqrnrvrtr1r1r2rp�s8z_chord.from_dictcKs||f|fSrer1)rVr�rur1r1r2r
�sz_chord._unpack_args�celery.chordc	s^|r|nd}|r|ndii}t�j||i|�t||�t||d�d��fd|i|��d|_dS)Nr1rurm)rVr�rnr&)ryrzrr(rl)rMrVr�r7rtrurnrvrr1r2rz�s

����z_chord.__init__cKs |jd|rd|inifi|��S)Nr1r�r�)rMr�rvr1r1r2r�sz_chord.__call__csxt|ttf�s0t|t�r0|��}|j|B|_|St|t�rht|j�dkrht|�}|��}|j|B|_|St	��
|�SdS)Nr))r?r%r�r!r�r�r*r+r3ryr�)rMr�rSrr1r2r�s�z
_chord.__or__cCs�t|jt�st|j|jd�|_|jj|||jd�}|jj|_d}|jr�|jj|||||d�}|}	t�}
|	r�|	j|
vr|td��|
�	|	j�|	j
dur�||	_
q�|	j
}	qf|S)Nrm)r�r�r&)r�r&rcr�zRecursive result parents)r?r+r%rnr�r�r�r��RuntimeError�addr�)rMr�rcr&r�r�r��
header_resultZbody_resultr��seenr1r1r2r�s.�
�

z
_chord.freezeFcKs"|j}t|t�r|j}d}|dur8|j|fi|��p6i}|j||fi|��}|j||fi|��t|t�r�|�tt	f||d�|���n&|�
�}|D]}|j||fi|��q�|dur�|j|fi|��|du�r|j
du�r|j|fi|��p�i}|j||fi|��}|j
j||fi|��dS)Nr2)r+r?r%rWr�r�rr3r�r=r�r9rXr�rY)rMr:r;r<r+r�Zstampsr7r1r1r2r9:s&

z_chord.stampc	KsR|r|nd}|r|ni}|r4|js4t|�t|j�n|j}|�dd�pN|jd}
t|jdfi|��}|
jfi|	��}
|�|
�}t|j	t
�r�|j	��nt
|j	||j�dt
��d�}|jj�r�t��*|j||f|
|d�|	��Wd�S1�s�0Y|	�rt|jfi|	��n|j}
|
�dd�}|du�r6|}|j||
|f||d�|
��S)Nr1r�rur�)rnr�)r�r�)r�ru)rwr|rtr�rur@r��_get_appr?r+r%rvrorr�r�rr�r�)rMrtrur�r�r�r�ZrouterZ
result_clsrvr�rnr+Zmerged_optionsZoption_task_idr1r1r2r�Ts6���
�

��&
z_chord.apply_asyncTcKsn|r|nd}|r|ni}|dur&|jn|}t|jt�r@|j��nt|j|jd�}|j|�||�j|d�fd�S)Nr1rm)�	propagate)rt)r�r?r+r%r�rnr�ro)rMrtrurSr�rvr+r1r1r2r�ns��z_chord.applycs�t|t�st|t�rt�|�}t|t�rNt|jd|j�}t�fdd�|D��St|t�r�|jddd�D]}��	|�}|dkrh|SqhdSt|t
�r���	|j�St|t�r�dSt|�S)z�Count the number of tasks in the given signature recursively.

        Descend into the signature object and return the amount of tasks it contains.
        r+c3s|]}��|�VqdSre�r;r��rir1r2r��r�z"_chord._descend.<locals>.<genexpr>r�Nrr))
r?r!r@rpr%r5r+�sumr�r;r&r�r*)riZsig_objZsubtasksZ	child_sigZ
child_sizer1rUr2r;ys 






z_chord._descendcs&t�jd�j�}t�fdd�|D��S)z@Return the number of tasks in this chord's header (recursively).r+c3s|]}��|�VqdSrerTr�r�r1r2r��r�z)_chord.__length_hint__.<locals>.<genexpr>)r5r+rVrGr1r�r2r-�sz_chord.__length_hint__r)cKs�|p|�|�}|j�d�pt�}|j�d�}
|rBt|jfi|��n|j}|rd|�dd�|j�|�|j|	|
d�}|�dd�|�dd�|�dd�|j|||
d�}|j	r�|j
j|||||d�|j||
fd|i|��}n|�
g�|jj|�}||_|S)	a�Execute 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).
        r�r�Nr�r"r&)rcr&r�)�intervalr!�max_retries)rRrvrorr@r�r�r�rDr+r�Zapply_chordr�r�rnrr�)rMrVr�r�rnrWr!rX�eagerr�rurvrcr�ZbodyresZheader_result_argsrPr1r1r2r��s2�
z
_chord.runc	sJt�j|i|��}zt|jddd�|jd<WnttfyDYn0|S)Nr�Tr�)ryr�r(rur.r�r�rr1r2r��s
�z_chord.clonecCs|j�|�|S)z(Links a callback to the chord body only.)r�r�r�r1r1r2r��sz_chord.linkcCs~t|�}|jjjr:t|j�pgD]}|�|jdd��q nt�	dt
�t|j�pRgD]}t|t�rT|�|�qT|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.
        Tr�aKtask_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.)r(rnr�Z#task_allow_error_cb_on_chord_headerrr+r�r��warnings�warnrr?r&r�)rMr\r7r1r1r2r��s	
�

z_chord.link_errorcCs|jD]}|�|�qdS)z�Sets 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.
        Nr0r1r1r1r2r�	s	
z_chord.set_immutablec	Cs�|jrtt|jt�rVt|jjddd�|jjd�|j�t|jjdd�|jd���Sdt|jd|j�|j��Sd|j�d�S)	Nrr7z%({} | {!r})r)rm�%z<chord without body: �>)	r�r?r�rr+�formatrr"rxr�r1r1r2r�	s���z_chord.__repr__cCs|�|j�Sre)rRr�r�r1r1r2rn*	sz
_chord.appcCsh|j}|durXz|jj}Wnty2|j}Yn0|rB|dj}|durX|durX|j}|durd|StSr)rxr+r.r)rMr�rnr+r1r1r2rR.	s
z_chord._get_appz
kwargs.headerzTasks in chord header.zkwargs.bodyzBody task of chord.)N)NN)NrMNNN)N)NNNNNN)NF)NNNNNNNN)NNTN)NNr)NFNN)N)r^r_r`rar�rp�staticmethodr
rzr�r�r�r9r�r�r;r-r�r�r�r�r�r�r
rnrRrr+r�r�r1r1rr2r:�sL;�

�
!
�
�

�
=
'


r:cOsL|�d�}t|t�r6t|tj�r(|��Stj||d�St|g|�Ri|��S)z�Create 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.
    rnrm)ror?r@rr�r�r!rp)Zvariesrtrurnr1r1r2r'D	s	

r'FcCsF|durBt|tj�r"|r4|��}nt|t�r4t|�}|durB||_|S)a�Ensure 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)r?rr�r�r@r'rx)rqrnr�r1r1r2r(X	s

r()T)NF)Orar�r�rZ�abcrr�collectionsr�collections.abcrr�r�	functoolsrr�rr	�typesr
Zkombu.utils.functionalrrZkombu.utils.objectsr
Zkombu.utils.uuidrZvinerZ
celery._staterZcelery.exceptionsrZ
celery.resultrrZcelery.utilsrZcelery.utils.collectionsrZcelery.utils.functionalrrrrrrrrZcelery.utils.objectsrZcelery.utils.textrr �__all__r3r8r=rArJr��registerr@r!r�rkr�r"r	r#r$rr%r:r&r'Zsubtaskr(Z
maybe_subtaskr1r1r1r2�<module>s�	
-q9
z?
&Z
&