File: //usr/local/lib/python3.9/site-packages/celery/app/__pycache__/amqp.cpython-39.pyc
a
X>h \ � @ s
d Z ddlZddlmZ ddlmZ ddlmZ ddlm Z ddl
m Z m Z m
Z
mZmZmZ ddlmZ dd lmZ dd
lmZ dd lmZ dd lmZ dd
lm Z ddl m Z ddl m!Z! ddl"m#Z$ dZ%dZ&dZ'edd�Z(d dd�Z)G dd� de*�Z+G d d � d �Z,dS ) z/Sending/Receiving Messages (Kombu integration).� N)�
namedtuple)�Mapping)� timedelta)�WeakValueDictionary)�
Connection�Consumer�Exchange�Producer�Queue�pools)� Broadcast)�
maybe_list)�cached_property)�signals)�
anon_nodename)�saferepr)�indent)�maybe_make_aware� )�routes)�AMQP�Queues� task_messagei �zS
.> {0.name:<16} exchange={0.exchange.name}({0.exchange.type}) key={0.routing_key}
r ��headers�
properties�body�
sent_event�utf-8c s � fdd�| � � D �S )Nc s* i | ]"\}}t |t�r |�� �n||�qS � )�
isinstance�bytes�decode)�.0�k�v��encodingr �9/usr/local/lib/python3.9/site-packages/celery/app/amqp.py�
<dictcomp>% s �z utf8dict.<locals>.<dictcomp>)�items)�dr' r r& r( �utf8dict$ s
�r, c s� e Zd ZdZdZd!� fdd� Z� fdd�Z� fdd �Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z d"dd�Z
dd� Zdd� Zdd � Zd d � Zed d � �Z� ZS )#r u� Queue name⇒ declaration mapping.
Arguments:
queues (Iterable): Initial list/tuple or dict of queues.
create_missing (bool): By default any unknown queues will be
added automatically, but if this flag is disabled the occurrence
of unknown queues in `wanted` will raise :exc:`KeyError`.
max_priority (int): Default x-max-priority for queues with none set.
NTc s� t � �� t� | _|| _|| _|| _|d u r0tn|| _|| _ |d ur\t
|t �s\dd� |D �}|pbi }|� � D ].\}}t
|t
�r�| �|�n| j|fi |�� qld S )Nc S s i | ] }|j |�qS r )�name)r# �qr r r( r) C � z#Queues.__init__.<locals>.<dictcomp>)�super�__init__r �aliases�default_exchange�default_routing_key�create_missingr � autoexchange� max_priorityr r r* r
�add�
add_compat) �self�queuesr3 r5 r6 r7 r4 r- r. �� __class__r r( r1 8 s
zQueues.__init__c s0 z | j | W S ty* t� �|� Y S 0 d S �N)r2 �KeyErrorr0 � __getitem__�r: r- r< r r( r@ H s zQueues.__getitem__c s8 | j r|js| j |_t� �||� |jr4|| j|j< d S r> )r3 �exchanger0 � __setitem__�aliasr2 )r: r- �queuer< r r( rC N s
zQueues.__setitem__c C s" | j r| �| �|��S t|��d S r> )r5 r8 � new_missingr? rA r r r( � __missing__U s zQueues.__missing__c K s& t |t�s | j|fi |��S | �|�S )a� Add new queue.
The first argument can either be a :class:`kombu.Queue` instance,
or the name of a queue. If the former the rest of the keyword
arguments are ignored, and options are simply taken from the queue
instance.
Arguments:
queue (kombu.Queue, str): Queue to add.
exchange (kombu.Exchange, str):
if queue is str, specifies exchange name.
routing_key (str): if queue is str, specifies binding key.
exchange_type (str): if queue is str, specifies type of exchange.
**options (Any): Additional declaration options used when
queue is a str.
)r r
r9 �_add)r: rE �kwargsr r r( r8 Z s
z
Queues.addc K s> |� d|�d�� |d d u r&||d< | �tj|fi |���S )N� routing_keyZ binding_key)�
setdefault�getrH r
� from_dict)r: r- �optionsr r r( r9 o s zQueues.add_compatc C s` |j d u s|j jdkr | j|_ |js,| j|_| jd urR|jd u rFi |_| �|j� || |j< |S )N� )rB r- r3 rJ r4 r7 �queue_arguments�_set_max_priority)r: rE r r r( rH v s
z Queues._addc C s& d|vr"| j d ur"|�d| j i�S d S )Nzx-max-priority)r7 �update)r: �argsr r r( rQ � s zQueues._set_max_priorityr c C s\ | j }|sdS dd� t|�� �D �}|r8td�|�|�S |d d td�|dd� �|� S )z/Format routing table into string for log dumps.rO c S s g | ]\}}t �� �|��qS r )� QUEUE_FORMAT�strip�format)r# �_r. r r r( �
<listcomp>� s �z!Queues.format.<locals>.<listcomp>�
r r N)� consume_from�sortedr* �
textindent�join)r: r � indent_first�active�infor r r( rV � s
�z
Queues.formatc K s, | j |fi |��}| jdur(|| j|j< |S )z�Add new task queue that'll be consumed from.
The queue will be active even when a subset has been selected
using the :option:`celery worker -Q` option.
N)r8 �
_consume_fromr- )r: rE rI r. r r r( �
select_add� s
zQueues.select_addc s |r � fdd�t |�D �� _dS )z�Select a subset of currently defined queues to consume from.
Arguments:
include (Sequence[str], str): Names of queues to consume from.
c s i | ]}|� | �qS r r )r# r- �r: r r( r) � s z!Queues.select.<locals>.<dictcomp>N)r
ra )r: �includer rc r( �select� s
�z
Queues.selectc sJ � rFt � �� | jdu r.| �� fdd�| D ��S � D ]}| j�|d� q2dS )z�Deselect queues so that they won't be consumed from.
Arguments:
exclude (Sequence[str], str): Names of queues to avoid
consuming from.
Nc 3 s | ]}|� vr|V qd S r> r )r# r$ ��excluder r( � <genexpr>� r/ z"Queues.deselect.<locals>.<genexpr>)r
ra re �pop)r: rg rE r rf r( �deselect� s
zQueues.deselectc C s t || �|�|�S r> )r
r6 rA r r r( rF � s zQueues.new_missingc C s | j d ur| j S | S r> )ra rc r r r( rZ � s
zQueues.consume_from)NNTNNN)r T)�__name__�
__module__� __qualname__�__doc__ra r1 r@ rC rG r8 r9 rH rQ rV rb re rj rF �propertyrZ �
__classcell__r r r< r( r ) s( �
r c @ s* e Zd ZdZeZeZeZeZeZ dZ
dZ dZ dZ
dZdd� Zedd� �Zedd � �Zd0d
d �Zd1d d
�Zdd� Zd2dd�Zd3dd�Zd4dd�Zdd� Zdd� Zed d � �Zed d � �Zej d d � �Ze d!d"� �Z ed#d$� �Z e j d%d$� �Z e d&d'� �Z e Z!ed(d)� �Z"ed*d+� �Z#ed,d-� �Z$d.d/� Z%dS )5r zApp AMQP API: app.amqp.Ni c C s* || _ | j| jd�| _| j j�| j� d S )N)r � )�app�
as_task_v1�
as_task_v2�task_protocolsZ_confZbind_to�_handle_conf_update)r: rr r r r( r1 � s
�z
AMQP.__init__c C s | j | jjj S r> )ru rr �confZ
task_protocolrc r r r( �create_task_message� s zAMQP.create_task_messagec C s | � � S r> )�_create_task_senderrc r r r( �send_task_message� s zAMQP.send_task_messagec C s� | j j}|j}|d u r |j}|d u r*|j}|s`|jr`d }|jdkrJddi}t|j| j||d�f}|d u rn| j n|}| �
|| j||||�S )NZquorumz x-queue-type)rB rJ rP ) rr rw �task_default_routing_key�task_create_missing_queuesZtask_queue_max_priority�task_default_queueZtask_default_queue_typer
r3 r6 �
queues_cls)r: r; r5 r6 r7 rw r4 rP r r r( r � s,
���z AMQP.Queuesc C s& t j| j|p| j| j�d|�| jd�S )z Return the current task router.r| )rr )�_routes�Routerr r; rr Zeither)r: r; r5 r r r( r�
s ��z AMQP.Routerc C s t �| jjj�| _d S r> )r �preparerr rw � task_routes�_rtablerc r r r( � flush_routes s zAMQP.flush_routesc K s: |d u r| j jj}| j|f||p.t| jj�� �d�|��S )N)�acceptr; )rr rw Zaccept_contentr �listr; rZ �values)r: Zchannelr; r� �kwr r r( � TaskConsumer s
���zAMQP.TaskConsumerr Fc ! s� |pd}|pi }t |ttf�s&td��t |t�s8td��|rx| �|d� |pT| j�� }|p`| jj}t |t
|d� |d�}t | t j �r�| �| d� |p�| j�� }|p�| jj}t |t
| d� |d�} t |t
�s�|o�|�� }t | t
�s�| o�| �� } |d u �rt|| j�}|d u �rt|| j�}|�s"|}� fdd �| �p4g D �} d
||||| |||
||g|||||�pdt� || | | d �} t| ||�p�d d
�||| |
|| d�f|�r�|||||||
|| d� nd d�S )Nr �!task args must be a list or tuple�(task keyword arguments must be a mapping� countdown��seconds)�tz�expiresc s i | ]}|� | �qS r r )r# �header�rN r r( r) H r/ z#AMQP.as_task_v2.<locals>.<dictcomp>�py)�lang�task�id�shadow�etar� �group� group_index�retries� timelimit�root_id� parent_id�argsrepr�
kwargsrepr�origin�
ignore_result�replaced_task_nesting�stamped_headers�stampsrO �Zcorrelation_id�reply_to)� callbacks�errbacks�chain�chord) �uuidr� r� r- rS rI r� r� r� r )r r� �tuple� TypeErrorr �_verify_secondsrr �now�timezoner r �numbers�Real�str� isoformatr �argsrepr_maxsize�kwargsrepr_maxsizer r )!r: �task_idr- rS rI r� r� �group_idr� r� r� r� r� r� r� �
time_limit�soft_time_limit�create_sent_eventr� r� r� r� r� r� r� r� r� r� r� r� rN r� r r r� r( rt s�
� �
������
�zAMQP.as_task_v2c K s |pd}|pi }| j }t|ttf�s,td��t|t�s>td��|rj| �|d� |pZ| j�� }|t |d� }t| t
j �r�| �| d� |p�| j�� }|t | d� } |o�|� � }| o�| � � } t
i ||p�dd�|||||||
|| || |
||f|| d �|�r||t|�t|�|
|| d
�nd d �S ) Nr r� r� r� r� r� rO r� )r� r� rS rI r� r� r� r� r� �utcr� r� r� �tasksetr� )r� r- rS rI r� r� r� r )r� r r� r� r� r r� rr r� r r� r� r� r r )r: r� r- rS rI r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� Z
compat_kwargsr� r r r( rs z sd
�����zAMQP.as_task_v1c C s |t k rt|� d|����|S )Nz is out of range: )�INT_MIN�
ValueError)r: �s�whatr r r( r� � s zAMQP._verify_secondsc s� | j jj�| j jj�| j jj�| j�| j� tjj �
tjj
�tj j � tj j
� tj j �tj j
�| j
�| j�| j jj� | j jj�
| j jj�d� ��������� �
� � �
��fdd� }|S )Nc s� |d u r �n|}|\}}}}|
r*|� |
� |r8|� |� |}|d u rP|d u rP�}|d urxt|t�rr|� | }}n|j}|
d u r�z |jj}
W n ty� Y n0 |
p��}
|d u r�z |jj}W n ty� d}Y n0 |r�|s�|dkr�d| }}n*|d u �r|jj�p�}|�p|j�p� }| d u �r<|�r<t|t ��s<|g} |d u �rJ�n|}|�rdt
�fi |��n�}��r��
||||| |||d� | j |f||| �p��
| �p��|||
| |||d� |��}� �rԈ |||||d� ��r<t|t ��r�||d ||d |d |d |d
d � n*�||d ||d |d
|d |d d � |�r�|�pJ�}|}t|t
��rb|j}|� |||d�� |j d|| ||d� |S )N�directrO )�senderr rB rJ �declarer r � retry_policy) rB rJ �
serializer� compression�retryr� �
delivery_moder� r �timeout�confirm_timeout)r� r r rB rJ r� r r r� r� )r� r� r� rS rI r� r� rS rI r� )rE rB rJ z task-sent)r� r� )rR r r� r- rB r� �AttributeError�typerJ r �dict�publishr� r )Zproducerr- �messagerB rJ rE Zevent_dispatcherr� r� r� r� r� r� r Z
exchange_typer� r� rI Zheaders2r r r �qnameZ_rp�retZevdZexname�Zafter_receiversZbefore_receiversZdefault_compressorZdefault_delivery_modeZ default_evdr3 Zdefault_policy�
default_queueZ
default_retryZ default_rkeyZdefault_serializerr; Zsend_after_publishZsend_before_publishZsend_task_sentZsent_receiversr r( rz � s�
����
� �
�
�
��z3AMQP._create_task_sender.<locals>.send_task_message)NNNNNNNNNNNNNN)rr rw Ztask_publish_retryZtask_publish_retry_policyZtask_default_delivery_moder� r; r Zbefore_task_publish�sendZ receiversZafter_task_publishZ task_sent�_event_dispatcherr3 r{ Ztask_serializerZtask_compression)r: rz r r� r( ry � s2
�,dzAMQP._create_task_senderc C s | j | jjj S r> )r; rr rw r} rc r r r( r� 6 s zAMQP.default_queuec C s | � | jjj�S )u" Queue name⇒ declaration mapping.)r rr rw Z task_queuesrc r r r( r; : s z AMQP.queuesc C s
| � |�S r> )r )r: r; r r r( r; ? s c C s | j d u r| �� | j S r> )r� r� rc r r r( r C s
z AMQP.routesc C s | � � S r> )r� rc r r r( �routerI s z AMQP.routerc C s |S r> r )r: �valuer r r( r� M s c C s0 | j d u r*tj| j�� | _ | jjj| j _| j S r> )�_producer_poolr Z producersrr Zconnection_for_write�pool�limitrc r r r( �
producer_poolQ s
�zAMQP.producer_poolc C s t | jjj| jjj�S r> )r rr rw Ztask_default_exchangeZtask_default_exchange_typerc r r r( r3 Z s
�zAMQP.default_exchangec C s
| j jjS r> )rr rw Z
enable_utcrc r r r( r� _ s zAMQP.utcc C s | j jjdd�S )NF)�enabled)rr �eventsZ
Dispatcherrc r r r( r� c s zAMQP._event_dispatcherc O s& d|v sd|v r"| � � | �� | _d S )Nr� )r� r� r� )r: rS rI r r r( rv i s
zAMQP._handle_conf_update)NNN)NN)NN)NNNNNNNr NNNNNNFNNNNNNNFNNNr )NNNNNNNr NNNNNNFNNNNN)&rk rl rm rn r r r ZBrokerConnectionr r~ r� r� r6 r� r� r1 r rx rz r� r� r� rt rs r� ry r� r; �setterro r r� r� Zpublisher_poolr3 r� r� rv r r r r( r � s�
�
�
^ �
<{
r )r )-rn r� � collectionsr �collections.abcr �datetimer �weakrefr Zkombur r r r r
r Z kombu.commonr Zkombu.utils.functionalr
Zkombu.utils.objectsr Zceleryr Zcelery.utils.nodenamesr Zcelery.utils.safereprr Zcelery.utils.textr r\ Zcelery.utils.timer rO r r �__all__r� rT r r, r� r r r r r r( �<module> s2 �