File: //usr/local/lib/python3.9/site-packages/celery/worker/__pycache__/strategy.cpython-39.pyc
a
X>h� � @ s� d Z ddlZddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z ddl
mZ dd lmZ dd
lmZ d d lmZ d d
lmZ dZee�Zdd� Zdd� Zej ej eej e efdd�Z dS )z'Task execution strategy (optimization).� N)� to_timestamp)�signals)�trace)�InvalidTaskError)�symbol_by_name)�
get_logger)�saferepr)�timezone� )�create_request_cls)�
task_reserved)�defaultc C s$ z$|� dd�|� di � }}|j W n2 ty> td��Y n tyV td��Y n0 |� d�|� d�|� d�|� d �|� d
�|� d �|� d �|� d
�|� d�|� d�|� dd�|� dd�|� d�|� d�|� d�d�}|�| jp�i � |� d�|� d�|� d�dd �}|||f|d |� d d �fS ) zECreate a fresh protocol 2 message from a hybrid protocol 1/2 message.�args� �kwargs�!Message does not have args/kwargs�(Task keyword arguments must be a mapping�lang�task�id�root_id� parent_id�group�meth�shadow�eta�expires�retriesr � timelimit)NN�argsrepr�
kwargsrepr�origin)r r r r r r r r r r r r r r r! � callbacks�errbacks�chordN�r" r# r$ �chainT�utc)�get�items�KeyErrorr �AttributeError�update�headers)�message�bodyr r r- �embedr r �@/usr/local/lib/python3.9/site-packages/celery/worker/strategy.py�hybrid_to_proto2 s@
�
��r2 c C s� z$|� dd�|� di � }}|j W n2 ty> td��Y n tyV td��Y n0 |jt|�t|�| jd� z|d |d< W n ty� Y n0 |� d �|� d
�|� d �d d
�}|||f|d|� dd�fS )z�Convert Task message protocol 1 arguments to protocol 2.
Returns:
Tuple: of ``(body, headers, already_decoded_status, utc)``
r r r r r )r r r- Ztasksetr r" r# r$ Nr% Tr' )r( r) r* r r+ r, r r- )r. r/ r r r0 r r r1 �proto1_to_proto2B s0
�
� �r3 c s� �j � �j�t�tj���j��o&�j} �o0�j�| o:�j ��j
j ��j ��j
��jj� �j�
�j��j�
t�j�}
t|
��j� ��d�� �jjj�tf� ��������� �
� � �
����������fdd� ��S )z�Default task execution strategy.
Note:
Strategies are here as an optimization, so sadly
it's not very easy to override.
)�appc
sb |d u r0d| j vr0| j| jd��� f\}}}}n2d| j v rPt| | j �\}}}}n�| |�\}}}}�| ||�� � ��||||d� � �r�� j� j� j� j� j d�} �
t
j | d| id� � j sĈ j�v rЈ �
� r�d S tjj�� d� ��r4�d� j� j� j� j� j� j� j�d d
�� j �o � j �� � j �o.� j �� d �
d }
d } � j �r�z*� j�r^|�� j ��} n|� j �j�} W nR ttf�y� } z4�d � j | � jd
d�d
d� � jdd� W Y d } ~ n
d } ~ 0 0 ��r҈
�j�}
| �r�|
�r��j � � �| �� |
dfdd�S | �r$�j � � �| �� fdd� �S |
�r6�� |
d�S �� � |�rV� fdd�|D � � � � d S )Nr F) Zon_ackZ on_rejectr4 �hostname�eventerr �connection_errorsr/ r- �decodedr' )r �namer r r �data)�extra)Zsender�requestz
task-receivedr r ) �uuidr9 r r r r r r r z2Couldn't convert ETA %r to timestamp: %r. Task: %rT)�safe)�exc_info)Zrequeuer
� )�priorityc s g | ] }|� ��qS r r )�.0�callback��reqr r1 �
<listcomp>� � z9default.<locals>.task_message_handler.<locals>.<listcomp>) �payloadr/ r- Zuses_utc_timezoner2 r r9 r r r �
_app_traceZ LOG_RECEIVEDr �revokedr Z
task_received�sendr r Z request_dictr( � isoformatr' r �
OverflowError�
ValueError�info�rejectZqosZincrement_eventually)
r. r/ ZackrP r" r r- r8 r' �contextZbucketr �exc�ZReqZ
_does_infor4 �apply_eta_task�call_atr7 �consumer�errorr6 Z
get_bucket�handler5 rO Zlimit_post_etaZ
limit_taskr3 Zrate_limits_enabledZ
revoked_tasksZ
send_eventr �task_message_handlerr Ztask_sends_events� to_system_tzrD r1 rY � s� �
��� �
�"
�
z%default.<locals>.task_message_handler)r5 r7 �logger� isEnabledFor�logging�INFOZevent_dispatcher�enabledrK Z send_events�timerrU rT Zdisable_rate_limitsZ task_buckets� __getitem__Zon_task_requestZ _limit_taskZ_limit_post_etar �Requestr �pool�
controller�staterJ r ) r r4 rV rO rW r rZ �bytesr3 �eventsrb r rS r1 r
c s(
�<Lr
)!�__doc__r] Zkombu.asynchronous.timerr Zceleryr Z
celery.appr rI Zcelery.exceptionsr Zcelery.utils.importsr Zcelery.utils.logr Zcelery.utils.safereprr Zcelery.utils.timer r< r re r �__all__�__name__r[ r2 r3 rO rW Z to_systemrf r
r r r r1 �<module> s& )"
�