Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/celery/__pycache__/canvas.cpython-39.pyc
Ðазад
a t�h�y � @ s~ d Z ddlZddlZddlZddlmZmZ ddlmZ ddl m Z ddl m Z ddl mZ ddl mZ dd lmZ dd lmZ dd lmZmZ dd lmZ dd lmZ ddlm Z ddl m Z ddl m 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/m0Z0 ddl1m2Z2 ddl3m4Z4m5Z5 dZ6dd� Z7d d � Z8d d � Z9dCd!d"�Z:G d#d$� d$ed%�Z;e%j<j=G d&d'� d'e>��Z?d(d)� Z@e?jAd*d+�G d,d-� d-e?��ZBG d.d*� d*eB�ZCG d/d0� d0e?�ZDe?�A� G d1d2� d2eD��ZEe?�A� G d3d4� d4eD��ZFe?�A� G d5d6� d6e?��Z*d7d8� ZGe?�A� G d9d:� d:e?��ZHe?jAd;d+�G d<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_signaturec C s� zt | j�}W n^ tyl z| j�� }W n ttfyH | Y Y S 0 |dkr`t| j�d n| Y S Y n0 |dkr�| jd S | S dS )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.� r N)�len�tasks� TypeError�__length_hint__�AttributeError�list)r% �size� r1 �}/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/celery/canvas.py�maybe_unroll_group* s $r3 c C s t | d| �S )N�name)�getattr��taskr1 r1 r2 �task_name_from< s r8 c K s | 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�headersr1 r1 r2 �_stamp_regen_task@ s r= Tc C s� |sdS | � � D ]�\}}||v rt|t�r>t| | || � qt|tttf�r`|rX|gn|| |< t|| t�r�t| | t�r�| | �|| � q|r| | du r�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�valuer1 r1 r2 rA I s$ rA c @ s� e Zd ZdZed�dd�Zdd�dd�Zed�dd �Zdd�d d �Ze ed�d d ��Z ed�dd�Z dd�dd�Z ed�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. )�returnc K s i S )a Method 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< r1 r1 r2 �on_group_start| s z StampingVisitor.on_group_startNc K s dS )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. Nr1 rL r1 r1 r2 � on_group_end� s z StampingVisitor.on_group_endc K s i S )a Method 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 �rM r" r< r1 r1 r2 �on_chain_start� s z StampingVisitor.on_chain_startc K s dS )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. Nr1 rP r1 r1 r2 � on_chain_end� s z StampingVisitor.on_chain_endc K s dS )a Method 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< r1 r1 r2 � on_signature� s z StampingVisitor.on_signaturec K s, t |jt�st|j�|_| j|jfi |��S )u Method 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 �rM rS �headerr1 r1 r2 �on_chord_header_start� s z%StampingVisitor.on_chord_header_startc K s | 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)rO r+ rU r1 r1 r2 �on_chord_header_end� s z#StampingVisitor.on_chord_header_endc K s i S )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. r1 rU r1 r1 r2 � on_chord_body� s z StampingVisitor.on_chord_bodyc K s i S )a Method 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 �callbackrV r1 r1 r2 � on_callback� s zStampingVisitor.on_callbackc K s i S )a Method 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 �errbackrV r1 r1 r2 � on_errback� s zStampingVisitor.on_errback)�__name__� __module__� __qualname__�__doc__r@ rN rO rQ rR r rT rW rX rY r[ r] r1 r1 r1 r2 rJ v s rJ )� metaclassc s� e Zd ZdZi Zd ZZddhZededd��Z edfdd��Z dg� fd d � Z d d � Z dd� Z dhdd�Zdidd�Zdjdd�Zdkdd�ZeZdldd�ZeZdmdd�Zdnd d �Zd d � Zdod!d"�Zdpd#d$�Zdqd%d&�Zd'd(� Zd)d*� Z d+d,� Z d-d.� Z d/d0� Z d1d2� 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_headersc s � �fdd�}|S )z�Register a new type of signature. Used as a class decorator, for example: >>> @Signature.register_type() >>> class mysig(Signature): >>> pass c s | � j �p| j< | S �N)�TYPESr^ )�subclass��clsr4 r1 r2 �_inner/ s z'Signature.register_type.<locals>._innerr1 )ri r4 rj r1 rh r2 � register_type' s zSignature.register_typec C s: |� d�}|r.| j| }|| ur.|j||d�S t||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_clsr1 r1 r2 rp 5 s zSignature.from_dictFc s~ || _ t|t�r t� �|� n\z |j} W n ty> |} Y n0 || _t� j| t|pVd�|p^i t|pfi fi | ��||d� d S )Nr1 �r7 �args�kwargs�optionsrl � immutable) �_appr? r@ �super�__init__r4 r. �_type�tuple) rM r7 rt ru rv �typerl rw rn �exZ task_name�� __class__r1 r2 rz B s �zSignature.__init__c O s$ | � ||d�\}}}| j|i |��S )z0Call the task directly (in the current process).N)�_merger} )rM � partial_args�partial_kwargsrt ru �_r1 r1 r2 �__call__Z s zSignature.__call__c O s | � ||�S )z5Shortcut to :meth:`apply_async` using star arguments.�� apply_async)rM r� r� r1 r1 r2 �delay_ s zSignature.delayc K sT |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. r1 c S s i | ]\}}|d ur||�qS re r1 ��.0�k�vr1 r1 r2 � <dictcomp>l � z#Signature.apply.<locals>.<dictcomp>)r> r� r} �apply)rM rt ru rv r1 r1 r2 r� c s zSignature.applyc K s� |r|nd}|r|ni }dd� |� � D �}z | j}W n tyH Y dS 0 |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:`~@Task.apply_async` and the :ref:`guide-calling` guide. r1 c S s i | ]\}}|d ur||�qS re r1 r� r1 r1 r2 r� � r� z)Signature.apply_async.<locals>.<dictcomp>N)r> � _apply_async� IndexErrorr� rt ru rv )rM rt ru � route_namerv �_applyr1 r1 r2 r� q s zSignature.apply_asyncc s� |r|nd}|r|ni }|durn�j � d�jv rJ�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 )aG Merge 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) r1 Nrd c s( i | ] \}}|� vs |�j vr||�qS r1 )rv r� �Zimmutable_optionsrM r1 r2 r� � s �z$Signature._merge.<locals>.<dictcomp>) �_IMMUTABLE_OPTIONSrv �union�setro r> rw rt ru r| r@ )rM rt ru rv �force� new_optionsr1 r� r2 r� � s � �zSignature._mergec K s� |r|nd}|r|ni }|s$|s$|r:| � |||�\}}}n| j| j| j }}}tj| jt|�|t|�| j | j d�| j d�}| j |_ |S )a7 Create 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. r1 rs rm ) r� rt ru rv r! rp r7 r| r rl rw rx r{ )rM rt ru �optsr' r1 r1 r2 �clone� s ��zSignature.clonec C s� | j }z |d }W n$ ty6 |p(t� }|d< Y n0 |rD||d< |rP||d< d|vrd| jj|d< |rxd|vrx||d< |r�||d< |dur�||d< | �|�S ) aC Finalize 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_torc r& N� group_index)rv �KeyErrorr rn Z thread_oid� AsyncResult) rM �_idrc r& r� r� r� r� �tidr1 r1 r2 �freeze� s$ zSignature.freezec C s6 | � � }|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� rt ru rv )rM rt ru rv r' r1 r1 r2 �replace s zSignature.replacec K s" |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)rM rw rv r1 r1 r2 r� s z Signature.setc C s || _ d S re �rw )rM rw r1 r1 r2 r� & s zSignature.set_immutableTc s� � � � � 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. rd N)rG c s" i | ]\}}|� d v r||�qS )rd r1 r� �r< r1 r2 r� ] r� z,Signature._stamp_headers.<locals>.<dictcomp>) �copyr/ �keysrA r� ro r� rv r> )rM �visitor_headersr; � self_headersr< rd Zredacted_optionsr1 r� r2 �_stamp_headers) s" zSignature._stamp_headersc K sb | j ||fi |�� |�� }d}|dur>|j| fi |��p<i }| j||fi |��}| jf i |��S )a Stamp 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� rT r� r� )rM r: r; r< r� r1 r1 r2 r9 e s zSignature.stampc K s |� � }d}t|�}t| j�d��p&g D ]`}t|| jd�}d}|dur\|j|fi |��pZi }| jf |||d�|��}|j ||fi |�� q(t|�}t| j�d��p�g D ]`}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�linkrm N)r� r; r� � link_error) r� r r rv ro r( rn r[ r� r9 r] )rM r: r; r< Znon_visitor_headersr� r� r� r1 r1 r2 r� { s< ����zSignature.stamp_linksc C s, | 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 )rM rH r> r1 r1 r2 �_with_list_option� s zSignature._with_list_optionc C s | � |�}||vr |�|� |S )zEAppends the given value to the list at the given key in self.options.)r� rF �rM rH rI r> r1 r1 r2 �append_to_list_option� s z Signature.append_to_list_optionc C s | � |�}|�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� rE r r� r1 r1 r2 �extend_list_option� s z Signature.extend_list_optionc C s | � 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� �rM rZ r1 r1 r2 r� � s zSignature.linkc C s | � 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� �rM r\ r1 r1 r2 r� � s zSignature.link_errorc C s | � |� | S )a2 Version 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� r1 r1 r2 �on_error� s zSignature.on_errorc C s6 t tj�t�| ggdd� t| j�d��p*g D ����S )zhReturn a recursive list of dependencies. "unchain" if you will, but with links intact. c s s | ]}|� � V qd S re )� flatten_links)r� r� r1 r1 r2 � <genexpr>� s �z*Signature.flatten_links.<locals>.<genexpr>r� )r/ � itertoolsr" � from_iterabler rv ro �rM r1 r1 r2 r� � s ��zSignature.flatten_linksc C sd t |t�r$tt| f|�� �| jd�S t |t�rFt|�}t| || jd�S t |t�r`t| || jd�S t 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_tasksrx r% r3 rn r! �NotImplemented�rM �otherr1 r1 r2 �__or__� s �� zSignature.__or__c C s | � |�S re )r� r� r1 r1 r2 �__ior__ s zSignature.__ior__c C s� | j }|j}| j�d�p t� }|�d ��N}|j�||�}|jj |d| j f d|i|��|j d� |� |�W d � S 1 s|0 Y d S )Nr� r7 )� connection) r} rn rv ro r �producer_or_acquire�backendZ on_task_call�control�electionr� r� r� )rM r} rn r� �producer�propsr1 r1 r2 r� s �zSignature.electionc O s( | j ||i dd�\}}}t| d ||�S )aA Return 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 )rM rt ru r� r1 r1 r2 r s zSignature.reprcallc C s | |t | �< t| �S re )�idr@ )rM �memor1 r1 r2 � __deepcopy__- s zSignature.__deepcopy__c C s | � � �� S re )r� ro r� r1 r1 r2 � __invert__1 s zSignature.__invert__c C s t t| �ffS re )r' r@ r� r1 r1 r2 � __reduce__4 s zSignature.__reduce__c C s t | �S re )r@ r� r1 r1 r2 �__json__9 s zSignature.__json__c C s | � � S re )r r� r1 r1 r2 �__repr__< s zSignature.__repr__c # s4 t � �� D ]$\}}t|t�r$|�� n||fV q d S re )ry r> r? �bytes�decode)rM r� r� r r1 r2 r> ? s zSignature.itemsc C s | j S re r6 r� r1 r1 r2 r4 C s zSignature.namec C s | j p| jj| d S �Nr7 )r{ rn r+ r� r1 r1 r2 r} H s zSignature.typec C s | j ptS re )rx r r� r1 r1 r2 rn L s z Signature.appc C s* z | j jW S ty$ | jj Y S 0 d S re )r} r� r� rn r� r1 r1 r2 r� P s zSignature.AsyncResultc C s4 z | j jW S ty. t| jj| d � Y S 0 d S r� )r} r� r� �_partialrn Z send_taskr� r1 r1 r2 r� W s zSignature._apply_asynczoptions.task_idz Task UUIDzoptions.parent_idzTask parent UUID.zoptions.root_idzTask root UUID.r7 z Name of task.rt z Positional arguments to task.ru zKeyword arguments to task.rv zTask execution options.rl zType of signaturerw z+Flag set if no longer accepts new arguments)N)N)NNNNNNFN)NN)NNN)NNNF)NN)NNNNNN)NNN)N)NFT)NF)F)>r^ r_ r` ra rf rx r{ r� � classmethodrk rp rz r� r� r� r� r� r� r r� �_freezer� r� r� r� r9 r� r� r� r� r� r� r� r� r� r� r� r r� r� r� r� r� r> �propertyr4 r r} rn r� r� r r� r� r� r7 rt ru rv rl rw � __classcell__r1 r1 r r2 r! � s� 8 � ! * � 1 < 1 �r! c C sJ |rt dd i| �S d| vr(t d|i| �S |d urFt d| d | i| �S d S )Nr" r )rv r+ �use_linkr1 r1 r2 �_prepare_chain_from_optionsj s )�r� r" )r4 c s� e Zd Zedd�Zed"dd��Z� fdd�Zdd � Zd d � Z � fd d �Z dd� Z d#dd�Z d$dd�Z d%dd�Zd&dd�Zddddddddejdf dd�Zd'd d �Zed d � �Zd d!� Z� ZS )(r� � kwargs.taskszTasks in chain.Nc sX |d d }|r@t |t�r.t|� }|d d< � fdd�|D �}| |fd� i|d ��S )Nru r+ c s g | ]}t |� d ��qS �rm �r( �r� r7 rm r1 r2 � <listcomp>� r� z$_chain.from_dict.<locals>.<listcomp>rn rv )r? r| r/ )ri rq rn r+ r1 rm r2 rp � s z_chain.from_dictc sb t |�dkr$t|d �r$t|d �n|}t� jddd|ifi |�� |�dd �| _d| _d | _d S )Nr) r z celery.chainr1 r+ r� r" ) r* r r ry rz �pop� _use_linkrl �_frozen�rM r+ rv r r1 r2 rz � s $� z_chain.__init__c O s | j r| �||�S d S re )r+ r� )rM rt ru r1 r1 r2 r� � s z_chain.__call__c C s� t |t�r�t|�}t |t�s&| �|�S | �� }|s6|S t |d t�rj|d j|B |d _t| �|| jd�S t| �t ||�| j d�S t |t �r�t| �t | �� |�� �| j d�S t |t ��r�| j�rJt | jd t��rJ| �� }t|jd || j d�|jd< t|j�dk�rFt |jd t��rF|jd j|jd B |jd _|jd d� |_|S | j�r�t | jd t��r�| �� }|jd j|B |jd _|S t| �t | �� |�| j d�S ntS d S )N���rm r) ���)r? r% r3 r� r� r& �bodyr} rn r rx r� r r! r+ r� r* r� )rM r� r+ rS r1 r1 r2 r� � sR �� �� � " ��z _chain.__or__c s: t �t� j|i |��}� �fdd�|jd D �|jd< |S )Nc s g | ]}�|� j d d��qS )T)rn r� )rx )r� rS �rM Z to_signaturer1 r2 r� � s �z _chain.clone.<locals>.<listcomp>r+ )r( ry r� ru �rM rt ru r' r r� r2 r� � s � z _chain.clonec C sj dd� | j D �}t| j�d��p"g D ]}|d �|� q$t| j�d��pJg D ]}|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. c S s g | ] }|� � �qS r1 �r� �r� �tr1 r1 r2 r� r� z(_chain.unchain_tasks.<locals>.<listcomp>r� r� r� )r+ r rv ro r� r� )rM r+ rS r7 r1 r1 r2 r� � s z_chain.unchain_tasksc K s� |r|nd}|r|ng }| j }|jjr`t� �$ | j||fi |��W d � S 1 sV0 Y | j||fd|i|r�t| jfi |��n| j��S )Nr1 rn )rn �conf�task_always_eagerr r� �runr@ rv )rM rt ru rv rn r1 r1 r2 r� s 2�z_chain.apply_asyncc K s� |r|nd}|r|ng }| p | j } | j}|du r@| jjdkr@d}|r\| js\t|�t| j� n| j}| j||| j| | || |||| d� \}}|r�|r�|d � d|� |� � }t |||�}|j f i |��}|s�|S |d S dS )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. r1 Nr) T)r� r r� ) rn r� r� � task_protocolrw r| rt � prepare_stepsr+ r� r� r� r� )rM rt ru rc r& r� r� r� � publisherr� r� r� rn r� rv r� r+ Zresults_from_prepareZ first_taskZresult_from_applyr1 r1 r2 r� s2 ���� z _chain.runc C s< | j | j| j| j||d | j|||d|d� \}}| _|d S )NF)r� r� r )r� rt ru r+ rn r� ) rM r� rc r& r� r� r� r� �resultsr1 r1 r2 r� G s �z _chain.freezeFc K s� d }|d ur"|j | fi |��p i }| j||fi |��}| j|fi |�� | jD ]}|j||fi |�� qN|d ur�|j| fi |�� d S re )rQ r� r� r+ r9 rR )rM r: r; r< r� r7 r1 r1 r2 r9 R s z _chain.stampTc C s: |p| j }| j}|du r(|jjdkr(d}t|�}|j}|j}d}d}g g }}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�}W n$ t�yB t||||d�}Y n0 |�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) Tr rm )r� r� r� rn )r� r� rn r� )r� rc r& r� �r� ) rn r� r� r� r r� rE r? r �CallableSignaturer% r3 r� r| rt r� r+ r& r� r. r� r� �parentr r� rF r� Zensure_chords_allowedr� ) rM rt ru r+ r� r� r� rn Z last_task_idrc Z chord_bodyr� rp r� r� ZstepsZ steps_popZ steps_extendZ prev_taskZprev_resr� �ir7 Z is_first_taskZ is_last_task�resr\ �noder1 r1 r2 r� _ s� % � � � z_chain.prepare_stepsc K s� |r|nd}|r|ni }d ||f }\}}| j D ]L}|�||�j|oL|�� ffi t| jfi |����}||d |_}\}}q0|S )Nr1 )NN)r+ r� r� ro r@ rv r� ) rM rt ru rv �lastZfargsZfkwargsr7 r� r1 r1 r2 r� s ��z _chain.applyc C s: | j }|d u r2z| jd j }W n ty0 Y n0 |p8tS �Nr )rx r+ � LookupErrorr �rM rn r1 r1 r2 rn s z _chain.appc C sJ | j s$dt| �j� dt| �d�d�S t| j d d d�dd � | j D ���S ) N�<�@z#xz: empty>r r7 z | c s s | ]}t |�V qd S re )�reprr� r1 r1 r2 r� r� z"_chain.__repr__.<locals>.<genexpr>)r+ r} r^ r� r �joinr� r1 r1 r2 r� s �z_chain.__repr__)N)NN) NNNNNNNNNNNNN)NNNNNN)NF)NN)r^ r_ r` r r+ r� rp rz r� r� r� r� r� r� r� r9 r! r� r� r� rn r� r� r1 r1 r r2 r� � s8 2 � 0 � � $ r� c s e Zd ZdZ� fdd�Z� ZS )r" a Chain tasks together. Each tasks follows one another, by being applied as a callback of the previous task. Note: If called with only one argument, then that argument must be an iterable of tasks to chain: this allows us to use generator expressions. Example: This is effectively :math:`((2 + 2) + 4)`: .. code-block:: pycon >>> res = chain(add.s(2, 2), add.s(4))() >>> res.get() 8 Calling a chain will return the result of the last task in the chain. You can get to the other tasks by following the ``result.parent``'s: .. code-block:: pycon >>> res.parent.get() 4 Using a generator expression: .. code-block:: pycon >>> lazy_chain = chain(add.s(i) for i in range(10)) >>> res = lazy_chain(3) Arguments: *tasks (Signature): List of task signatures to chain. If only one argument is passed and that argument is an iterable, then that'll be used as the list of signatures to chain instead. This means that you can use a generator expression. Returns: ~celery.chain: A lazy signature that can be called to apply the first task in the chain. When that task succeeds the next task in the chain is applied, and so on. c s� |s�|r�t |�dks t|d �r�t |�dkr4|d n|}ttj|t� �}| tkr|t|t�r|t|| �s|t� j | |j fi |��S |S t� j | g|�R i |��S )Nr) r ) r* r r �operator�or_r� r" r? ry �__new__r+ )ri r+ ru Z new_instancer r1 r2 r O s z chain.__new__)r^ r_ r` ra r r� r1 r1 r r2 r" s 0c sB e Zd ZdZedd�Zed dd��Z� fdd�Zd dd �Z � Z S ) �_basemapNr7 �itc C s | | � |d �d|i|d ��S �Nru rn rv �� _unpack_args�ri rq rn r1 r1 r2 rp a s z_basemap.from_dictc s, t � j| jd|t|�d�fddi|�� d S )Nr1 �r7 r rw T)ry rz � _task_namer )rM r7 r rv r r1 r2 rz e s ���z_basemap.__init__c K sX |r|nd}|r|ni }| � | j�\}}| jjd|t|�d�fdt| j�d��i|��S )Nr1 r r� r7 )r ru r} r� r/ r8 ro )rM rt ru r� r7 r r1 r1 r2 r� j s ���z_basemap.apply_async)N)NN) r^ r_ r` r r r r� rp rz r� r� r1 r1 r r2 r ] s r c @ s e Zd ZdZdZdd� ZdS )r# z�Map operation for tasks. Note: Tasks executed sequentially in process, this is not a parallel operation like :class:`group`. z celery.mapc C s. | � | j�\}}d|j� dtt|�d�� d�S )N�[z (x) for x in �d �]�r ru r7 r r �rM r7 r r1 r1 r2 r� � s z xmap.__repr__N�r^ r_ r` ra r r� r1 r1 r1 r2 r# u s r# c @ s e Zd ZdZdZdd� ZdS )r$ z.Map operation for tasks, using star arguments.zcelery.starmapc C s. | � | j�\}}d|j� dtt|�d�� d�S )Nr z(*x) for x in r r r r r1 r1 r2 r� � s zxstarmap.__repr__Nr r1 r1 r1 r2 r$ � s r$ c sb e Zd ZdZeddd�Zeddd��Z� fdd �Zd d � Z dd d �Z dd� Z eddd��Z � Z S )r z)Partition of tasks into chunks of size n.r7 r �nNc C s | | � |d �d|i|d ��S r r r r1 r1 r2 rp � s zchunks.from_dictc s, t � jdd|t|�|d�fddi|�� d S )Nz celery.chunksr1 )r7 r r rw T)ry rz r )rM r7 r r rv r r1 r2 rz � s ���zchunks.__init__c K s | j f i |��S re r� )rM rv r1 r1 r2 r� � s zchunks.__call__c K s@ |r|nd}|r|ni }| � � j||fdt| j�d��i|��S )Nr1 r� r7 )r% r� r8 ru ro )rM rt ru r� r1 r1 r2 r� � s ���zchunks.apply_asyncc s: � � � j�\�}}t� �fdd�tt|�|�D �� jd�S )Nc 3 s | ]}t �|� jd �V qdS �rm N)r$ rx )r� �part�rM r7 r1 r2 r� � s �z chunks.group.<locals>.<genexpr>rm )r ru r% �_chunks�iterrx )rM r r r1 r r2 r% � s ��z chunks.groupc C s | ||||d�� S )Nrm r1 )ri r7 r r rn r1 r1 r2 � apply_chunks� s zchunks.apply_chunks)N)NN)N)r^ r_ r` ra r r r� rp rz r� r� r% r r� r1 r1 r r2 r � s r c sx t | t�rt| � d�} t | ttf�r,| j} nHt | tj�r@| g} n4t | t�rbt � fdd�| D ��} n� fdd�| D �} | S )Nrm c 3 s | ]}t |� d �V qdS r �r' r� rm r1 r2 r� � r� z _maybe_group.<locals>.<genexpr>c s g | ]}t |� d ��qS r� r r� rm r1 r2 r� � r� z _maybe_group.<locals>.<listcomp>) r? r@ r' r% r� r+ r r� r r )r+ rn r1 rm r2 � _maybe_group� s r c s� e Zd ZdZedd�Zed4dd��Z� fdd�Zd d � Z d d � Z d5dd�Z d6dd�Z d7dd�Z dd� Zd8dd�Zdd� Zd d � Zejejeefd d �Zd9d d!�Zd"d#� Zd:d$d%�Zd;d&d'�ZeZ d(d)� Z d*d+� Z d,d-� Z d.d/� Z d0d1� Z!e"d2d3� �Z#� Z$S )<r% aq Creates a group of tasks to be executed in parallel. A group is lazy so you must call it to take action and evaluate the group. Note: If only one argument is passed, and that argument is an iterable then that'll be used as the list of tasks instead: this allows us to use ``group`` with generator expressions. Example: >>> lazy_group = group([add.s(2, 2), add.s(4, 4)]) >>> promise = lazy_group() # <-- evaluate: returns lazy result. >>> promise.get() # <-- will wait for the task to return [4, 8] Arguments: *tasks (List[Signature]): A list of signatures that this group will call. If there's only one argument, and that argument is an iterable, then that'll define the list of signatures instead. **options (Any): Execution options applied to all tasks in the group. Returns: ~celery.group: signature that when called will then call all of the tasks in the group (and return a :class:`GroupResult` instance that can be used to inspect the state of the group). r� zTasks in group.Nc sJ |d d }t |�� fdd�|D �� |d d< }| |fd� i|d ��S )a Create 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). ru r+ c 3 s | ]}t |� d �V qdS r r� r� rm r1 r2 r� s z"group.from_dict.<locals>.<genexpr>rn rv )r} )ri rq rn Z orig_tasksZ rebuilt_tasksr1 rm r2 rp � s 2 �zgroup.from_dictc sr t |�dkrL|d }t|t�r$|j}t|tj�r:|�� g}t|t�sLt|�}t � j ddd|ifi |�� d| _ d S )Nr) r z celery.groupr1 r+ r% ) r* r? r% r+ r r� r� r r ry rz rl r� r r1 r2 rz s zgroup.__init__c O s | j |fi |��S re r� )rM r� rv r1 r1 r2 r� / s zgroup.__call__c C s t | || jd�S )N)r� rn )r& rx r� r1 r1 r2 r� 2 s z group.__or__� �?c C s0 t |||dd�}| jD ]}|jt|�d� q| S )NT)Z repeatlast)� countdown)r r+ r� �next)rM �start�stop�stepr r7 r1 r1 r2 �skew6 s z group.skewTc K s |r|nd}|d ur t d��|d ur,t d��| j}|jjrN| j||fi |��S | js\| �� S | �|�\}} } | �| jg | | |�} t � } t | j | ||| f||d�|���} | jj | | | d�}| � � t|�dkr�t|d t �r�|d }|j}|r�|r�|�|� |S )Nr1 z%Cannot add link to group: use a chordz5Cannot add link to group: do that on individual tasks�rt ru )Z ready_barrierr) r )r, rn r� r� r� r+ r� � _freeze_gid� _preparedr r/ � _apply_tasksr �finalizer* r? Zcurrent_worker_taskZ add_trail)rM rt ru � add_to_parentr� r� r� rv rn rc r� r+ �pr� �resultZ parent_taskr1 r1 r2 r� = s8 ��� zgroup.apply_asyncc sn � r� nd� �r�ni �| j }| js,| �� S | ���\�}}| �| jg |||�}|�|� ��fdd�|D ��S )Nr1 c s( g | ] \}}}|j f � �d �����qS )r& )r� )r� rS r� �rt ru rv r1 r2 r� i s z group.apply.<locals>.<listcomp>)rn r+ r� r' r( r )rM rt ru rv rn rc r� r+ r1 r. r2 r� a s �z group.applyc C s | j D ]}|�|� qd S re �r+ r� �rM rw r7 r1 r1 r2 r� m s zgroup.set_immutableFc K s� d }|d ur"|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�|| jd d �< n|| _|d ur�|j| fi |�� d S )N�r: r; rm )rN r� r� r? r+ r �mapr� r= r( rn r9 rF r rO )rM r: r; r< r� � new_tasksr7 r1 r1 r2 r9 q s" z group.stampc C s |� � jdd�}| jd �|�S )NTr� r )r� r� r+ r� �rM rS r1 r1 r2 r� � s z group.linkc s t � �� t� fdd�| jD ��S )Nc 3 s | ]}|� � jd d��V qdS )Tr� N)r� r� )r� Z child_task�rS r1 r2 r� � r� z#group.link_error.<locals>.<genexpr>)r( r| r+ r4 r1 r5 r2 r� � s zgroup.link_errorc c s� t |�D ]�\} } || |�r$| �� } n || |d�} || t�rZ| �| j||||�} | E dH q|rx| jsx| |�| | j� | _| | j||| d�|fV qdS )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)``. rm N)rc r� r� )� enumerater� r% r( r+ rw rt r� ) rM r+ r� rc r� rn r� rp r? r| �indexr7 Zunrollr1 r1 r2 r( � s � zgroup._preparedc k s |p| j }|�|���}d} t�|�\} }t| d� t| d�} t|�D ]�\} }|\}}}|durb|n |j�d�}| t� |�7 } |dur�| du r�|j � || � |j f |d|||d�| �� |r�|j s�|js�| jd7 _|j|dd� t| d�} |V qDW d � n1 �s 0 Y dS ) aU Run 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. r Nr& F)r� r+ r& rt ru r) T)Zweak)rn r� r� �teer! r6 rv ro �_chord�_descendr� Zset_chord_sizer� Z cancelled�readyr0 Zthen)rM r+ r� rn r, r+ r& rt ru rv Z chord_sizeZ tasks_shiftedZ next_taskZ task_indexZ current_taskrS r� rc Z chord_objr1 r1 r2 r) � s, �� zgroup._apply_tasksc sF i � j �� fdd�|�� D ��}|�dt� � |d< }|||�d�fS )z:Freeze the group id by the existing task_id or a new UUID.c s* i | ]"\}}|� j vs |� jvr||�qS r1 )r� rv r� r� r1 r2 r� s �z%group._freeze_gid.<locals>.<dictcomp>r� rc r� )rv r> r� r ro )rM rv rc r1 r� r2 r' s � � zgroup._freeze_gidc C s | j }z |d }W n$ ty6 |p(t� }|d< Y n0 |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� rc r& Nr� r� r� )rv r� r r� r? r+ r r� r8 � _unroll_tasksr � _freeze_tasksr/ �_freeze_unrollr ) rM r� rc r& r� r� r� r� �gidZtasks1Ztasks2r� r3 r1 r1 r2 �_freeze_group_tasks s2 � zgroup._freeze_group_tasksc C s | j j| j||||||d�� S )N)r� rc r& r� r� r� )rn r r@ )rM r� rc r& r� r� r� r1 r1 r2 r� O s �z group.freezec # s& � ���fdd�t |�D �E dH dS )zKCreates a generator for the AsyncResult of each task in the tasks argument.c 3 s& | ] \}}|j �� ��|d �V qdS )�rc r& r� r� r� N)r� )r� r� r7 �r& rc r� r� r1 r2 r� Z s ��z&group._freeze_tasks.<locals>.<genexpr>N)r6 )rM r+ rc r& r� r� r1 rB r2 r= X s �zgroup._freeze_tasksc # s � fdd�|D �E dH dS )z?Creates a generator for the cloned tasks of the tasks argument.c 3 s | ]}t |� jd ��� V qdS r )r( rx r� r� r� r1 r2 r� d r� z&group._unroll_tasks.<locals>.<genexpr>Nr1 �rM r+ r1 r� r2 r< a s zgroup._unroll_tasksc c sn t | 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. r rm rA r) N) r r+ r( �popleftrx r� r? r% � extendleftrF r� ) rM r3 rc r& r� r� �stackr� r7 r1 r1 r2 r> f s �zgroup._freeze_unrollc C s( | j r$t| j d d d| j �d��S dS )Nr r7 zgroup(�)zgroup(<empty>))r+ r r� r1 r1 r2 r� � s �zgroup.__repr__c C s t | j�S re )r* r+ r� r1 r1 r2 �__len__� s z group.__len__c C sB | j }|d u r2z| jd j}W n ty0 Y n0 |d ur>|S tS r� )rx r+ rn r� r r r1 r1 r2 rn � s z group.app)N)r Nr )NNTNNN)NN)NF)NNNNNNN)NNNNNN)NNNNNN)%r^ r_ r` ra r r+ r� rp rz r� r� r% r� r� r� r9 r� r� r r� r! r? r| r( r) r' r@ r� r� r= r< r> r� rH r� rn r� r1 r1 r r2 r% � sN 7 � $ � 1 � @ � 3 � $r% r& c s� e Zd ZdZed1dd��Zed2dd��Zd3� fdd � Zd4d d �Z � fd d �Z d5dd�Z d6dd�Z d7dd�Z d8dd�Zedd� �Zdd� Zd9d d �Z� fd d �Zd!d"� Zd#d$� Zd%d&� Zd'd(� Zed)d*� �Zd:d+d,�Zed-d.�Zed/d0�Z � Z S );r9 aL Barrier 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 Nc C s6 |� � }| jf i |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). ru rn )r� r )ri rq rn rv rt r1 r1 r2 rp � s 8 z_chord.from_dictc K s | |f|fS re r1 )rV r� ru r1 r1 r2 r � s z_chord._unpack_args� celery.chordc s^ |r|nd}|r|ndi i}t � j||i |�t||�t||d�d��fd|i|�� d| _d S )Nr1 ru rm )rV r� rn r& )ry rz r r( rl )rM rV r� r7 rt ru rn rv r r1 r2 rz � s ����z_chord.__init__c K s | j d|rd|ini fi |��S )Nr1 r� r� )rM r� rv r1 r1 r2 r� s z_chord.__call__c sx t |ttf�s0t |t�r0| �� }|j|B |_|S t |t�rht|j�dkrht|�}| �� }|j|B |_|S t � � |�S d S )Nr) ) r? r% r� r! r� r� r* r+ r3 ry r� )rM r� rS r r1 r2 r� s � z _chord.__or__c C s� t | jt�s t| j| jd�| _| jj||| jd�}| jj| _d }| jr�| jj|||||d�}|} t� } | r�| j| v r|td��| � | j� | j d u r�|| _ q�| j } qf|S )Nrm )r� r� r& )r� r& rc r� zRecursive result parents) r? r+ r% rn r� r� r� r� � RuntimeError�addr� ) rM r� rc r& r� r� r� � header_resultZ body_resultr� �seenr1 r1 r2 r� s. � � z _chord.freezeFc K s" | j }t|t�r|j }d }|d ur8|j| fi |��p6i }| j||fi |��}| j||fi |�� t|t�r�|�tt f||d�|��� n&|� � }|D ]}|j ||fi |�� q�|d ur�|j | fi |�� |d u�r | j d u�r |j| fi |��p�i }| j||fi |��}| j j ||fi |�� d S )Nr1 )r+ r? r% rW r� r� r r2 r� r= r� r9 rX r� rY )rM r: r; r< r+ r� Zstampsr7 r1 r1 r2 r9 < s& z _chord.stampc K sR |r|nd}|r|ni }|r4| j s4t|�t| j� n| j}|�dd �pN| jd } t| jd fi |��}| jf i | ��} | �| �} t| j t �r�| j �� nt | j | | j � dt � �d�} | jj� r�t� �* | j||f| |d�| ��W d � S 1 � s�0 Y | �rt| j fi | ��n| j } | �dd �}|d u �r6|}| j| | |f||d�| ��S )Nr1 r� ru r� )rn r� )r� r� )r� ru )rw r| rt r� ru r@ r� �_get_appr? r+ r% rv ro r r� r� r r� r� )rM rt ru r� r� r� r� ZrouterZ result_clsrv r� rn r+ Zmerged_optionsZoption_task_idr1 r1 r2 r� V s6 ��� � ��& z_chord.apply_asyncTc K sn |r|nd}|r|ni }|d u r&| j n|}t| jt�r@| j�� nt| j| jd�}|j|�||�j|d�fd�S )Nr1 rm )� propagate)rt )r� r? r+ r% r� rn r� ro )rM rt ru rO r� rv r+ r1 r1 r2 r� p s ��z _chord.applyc s� t |t�s t |t�r t�|�}t |t�rNt|jd|j�}t� fdd�|D ��S t |t�r�|jddd� D ] }� � |�}|dkrh| S qhdS t |t �r�� � |j �S t |t�r�dS t |�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+ c 3 s | ]}� � |�V qd S re �r: r� �ri r1 r2 r� � r� z"_chord._descend.<locals>.<genexpr>r� Nr r) ) r? r! r@ rp r% r5 r+ �sumr� r: r& r� r* )ri Zsig_objZsubtasksZ child_sigZ child_sizer1 rQ r2 r: { s z_chord._descendc s& t � jd� j�}t� fdd�|D ��S )z@Return the number of tasks in this chord's header (recursively).r+ c 3 s | ]}� � |�V qd S re rP r� r� r1 r2 r� � r� z)_chord.__length_hint__.<locals>.<genexpr>)r5 r+ rR rC r1 r� r2 r- � s z_chord.__length_hint__r) c K s� |p | � |�}|j�d�p t� } |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& )rc r& r� )�intervalr � max_retries)rN rv ro r r@ r� r� r� r@ r+ r� Z apply_chordr� r� rn r r� )rM rV r� r� rn rS r rT �eagerr� ru rv rc r� ZbodyresZheader_result_argsrL r1 r1 r2 r� � s2 � z _chord.runc sJ t � j|i |��}z t|jd dd�|jd< W n ttfyD Y n0 |S )Nr� Tr� )ry r� r( ru r. r� r� r r1 r2 r� � s �z _chord.clonec C s | j �|� |S )z(Links a callback to the chord body only.)r� r� r� r1 r1 r2 r� � s z _chord.linkc C sV t |�}| jjjr:t| j�p g D ]}|�|jdd�� q n t� dt � | j �|� |S )at Links 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� aK task_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( rn r� Z#task_allow_error_cb_on_chord_headerr r+ r� r� �warnings�warnr r� )rM r\ r7 r1 r1 r2 r� � s � z_chord.link_errorc C s | j D ]}|�|� 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. Nr/ r0 r1 r1 r2 r� s z_chord.set_immutablec C s� | j rtt| j t�rVt| j jd d d�| j jd �| j�t| j jdd � | jd���S dt| j d | j �| j�� S d| j�d�S ) Nr r7 z %({} | {!r})r) rm �%z<chord without body: �>) r� r? r� r r+ �formatr r" rx r� r1 r1 r2 r� s ���z_chord.__repr__c C s | � | j�S re )rN r� r� r1 r1 r2 rn % s z _chord.appc C sh | j }|d u rXz | jj}W n ty2 | j}Y n0 |rB|d j }|d u rX|d urX|j }|d urd|S tS r� )rx r+ r. r )rM r� rn r+ r1 r1 r2 rN ) s z_chord._get_appz kwargs.headerzTasks in chord header.z kwargs.bodyzBody task of chord.)N)NN)NrI NNN)N)NNNNNN)NF)NNNNNNNN)NNTN)NNr) NFNN)N) r^ r_ r` ra r� rp � staticmethodr rz r� r� r� r9 r� r� r: r- r� r� r� r� r� r� r rn rN r r+ r� r� r1 r1 r r2 r9 � sL ; � � ! � � � = r9 c O sL |� d�}t| t�r6t| tj�r(| �� S tj| |d�S t| g|�R i |��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. rn rm )ro r? r@ r r� r� r! rp )Zvariesrt ru rn r1 r1 r2 r' ? s r' Fc C sF | 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? r r� r� r@ r' rx )rq rn r� r1 r1 r2 r( S s r( )T)NF)Ora r� r rV �abcr r � collectionsr �collections.abcr r� r � functoolsr r� r r �typesr Zkombu.utils.functionalr r Zkombu.utils.objectsr Zkombu.utils.uuidr Zviner Z celery._stater Zcelery.exceptionsr Z celery.resultr r Z celery.utilsr Zcelery.utils.collectionsr Zcelery.utils.functionalr r r r r r r r Zcelery.utils.objectsr Zcelery.utils.textr r �__all__r3 r8 r= rA rJ r� �registerr@ r! r� rk r� r" r r# r$ r r% r9 r& r' Zsubtaskr( Z maybe_subtaskr1 r1 r1 r2 �<module> s� -q 9 |? & Z
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка