Server : LiteSpeed System : Linux barito.iixcp.rumahweb.net 5.14.0-611.49.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Apr 21 16:39:08 EDT 2026 x86_64 User : elvh3918 ( 1528) PHP Version : 8.2.31 Disable Function : mail Directory : /usr/local/lib/python3.9/site-packages/celery/contrib/__pycache__/ |
a
��Sh8 � @ sd d Z ddlZddlmZ ddlmZmZ ddlmZm Z ddl
m Z ddl m
Z
ddlmZ dd lmZ dd
lmZ d Zd ZG d
d� de�ZG dd� d�Zd4dd�Zd5dd�Zdd� Zeddfdd�Z dd� Z d6dd �Z d d � Z d d � Z d!d"� Z!d#d$� Z"G d%d&� d&�Z#d7d)d*�Z$d+d,� Z%d-d.� Z&d/d0� Z'd1d2� Z(ee ed3�Z)ee%ed3�Z*ee&ed3�Z+ee'ed3�Z,dS )8z,Message migration tools (Broker <-> Broker).� N)�partial)�cycle�islice)�Queue� eventloop)�
maybe_declare)� ensure_bytes)�app_or_default)�
worker_direct)� str_to_list)�
StopFiltering�State� republish� migrate_task�
migrate_tasks�move�
task_id_eq�
task_id_in� start_filter�move_task_by_id�
move_by_idmap�move_by_taskmap� move_direct�move_direct_by_idzGMoving task {state.filtered}/{state.strtotal}: {body[task]}[{body[id]}]c @ s e Zd ZdZdS )r z*Semi-predicate used to signal filter stop.N)�__name__�
__module__� __qualname__�__doc__� r r �@/usr/local/lib/python3.9/site-packages/celery/contrib/migrate.pyr s r c @ s0 e Zd ZdZdZdZdZedd� �Zdd� Z dS )r
zMigration progress state.r c C s | j s
dS t| j �S )N�?)� total_apx�str��selfr r r �strtotal& s zState.strtotalc C s$ | j rd| j � �S | j� d| j� �S )N�^�/)�filtered�countr% r# r r r �__repr__, s zState.__repr__N)
r r r r r) r( r! �propertyr% r* r r r r r
s
r
c
C s� |s g d�}t |j�}|j|j|j }}}|du r<|d n|}|du rP|d n|}|j|j } }
|�dd�} |�dd�} | dur�t| �nd} |D ]}
|�|
d� q�| j t |�f||| || |
| d�|�� dS )zRepublish message.)Zapplication_headers� content_type�content_encoding�headersN�exchange� routing_key� compression�
expiration)r/ r0 r1 r. r, r- r2 )
r �body�
delivery_infor. �
propertiesr, r- �pop�float�publish)�producer�messager/ r0 Z remove_propsr3 �infor. �props�ctype�encr1 r2 �keyr r r r 2 s*
�
��r c C s> |j }|du ri n|}t| ||�|d �|�|d �d� dS )z Migrate single task message.Nr/ r0 �r/ r0 )r4 r �get)r9 Zbody_r: �queuesr; r r r r P s �r c s � �fdd�}|S )Nc s �r| d �vrd S � | |�S �N�taskr �r3 r: ��callback�tasksr r r( [ s z!filter_callback.<locals>.filteredr )rG rH r( r rF r �filter_callbackY s rI c sV t |�}t���|jj|dd�� t|� �d�}� �fdd�}t|| |f�|d�|��S )z)Migrate tasks from one broker to another.F)Z auto_declare�rB c sh | � j �}��| j| j�|_|j| jkr:��| j|j�|_|jj| jkr\��| j| j�|j_|�� d S �N)�channelrA �namer0 r/ Zdeclare)�queueZ new_queue�r9 rB r r �on_declare_queuek s
�z'migrate_tasks.<locals>.on_declare_queue)rB rP )r �prepare_queues�amqp�Producerr r )�source�destZmigrate�apprB �kwargsrP r rO r r c s
��r c C s t |t�r| jj| S |S rK )�
isinstancer" rR rB )rV �qr r r � _maybe_queuey s
rZ c
s� t � �� � fdd�|pg D �p d}
� j|dd��V�� j����t� ���������� f dd�} t� �| fd|
i| ��W d � S 1 s�0 Y dS ) aG Find tasks by filtering them and move the tasks to a new queue.
Arguments:
predicate (Callable): Filter function used to decide the messages
to move. Must accept the standard signature of ``(body, message)``
used by Kombu consumer callbacks. If the predicate wants the
message to be moved it must return either:
1) a tuple of ``(exchange, routing_key)``, or
2) a :class:`~kombu.entity.Queue` instance, or
3) any other true value means the specified
``exchange`` and ``routing_key`` arguments will be used.
connection (kombu.Connection): Custom connection to use.
source: List[Union[str, kombu.Queue]]: Optional list of source
queues to use instead of the default (queues
in :setting:`task_queues`). This list can also contain
:class:`~kombu.entity.Queue` instances.
exchange (str, kombu.Exchange): Default destination exchange.
routing_key (str): Default destination routing key.
limit (int): Limit number of messages to filter.
callback (Callable): Callback called after message moved,
with signature ``(state, body, message)``.
transform (Callable): Optional function to transform the return
value (destination) of the filter function.
Also supports the same keyword arguments as :func:`start_filter`.
To demonstrate, the :func:`move_task_by_id` operation can be implemented
like this:
.. code-block:: python
def is_wanted_task(body, message):
if body['id'] == wanted_id:
return Queue('foo', exchange=Exchange('foo'),
routing_key='foo')
move(is_wanted_task)
or with a transform:
.. code-block:: python
def transform(value):
if isinstance(value, str):
return Queue(value, Exchange(value), value)
return value
move(is_wanted_task, transform=transform)
Note:
The predicate may also return a tuple of ``(exchange, routing_key)``
to specify the destination to where the task should be moved,
or a :class:`~kombu.entity.Queue` instance.
Any other true value means that the task will be moved to the
default exchange/routing_key.
c s g | ]}t � |��qS r )rZ )�.0rN )rV r r �
<listcomp>� � zmove.<locals>.<listcomp>NF)�poolc s� �| |�}|r��r�|�}t |t�rBt|�j� |jj|j }}nt|���\}}t�|||d� |� � � j
d7 _
� r�� �| |� �r��j
�kr�t � �d S )Nr@ � ) rX r r Zdefault_channelr/ rM r0 � expand_destr �ackr( r )r3 r: �ret�ex�rk) rG �connr/ �limit� predicater9 r0 �state� transformr r �on_task� s"
� zmove.<locals>.on_task� consume_from)r Zconnection_or_acquirerR rS r
r ) rg �
connectionr/ r0 rT rV rG rf ri rW rB rj r )
rV rG re r/ rf rg r9 r0 rh ri r r s > r c C s6 z | \}}W n t tfy, || }}Y n0 ||fS rK )� TypeError�
ValueError)rb r/ r0 rc rd r r r r` � s
r` c C s |d | kS )z'Return true if task id equals task_id'.�idr )�task_idr3 r: r r r r � s r c C s |d | v S )z-Return true if task id is member of set ids'.ro r )�idsr3 r: r r r r � s r c C s@ t | t�r| �d�} t | t�r0tdd� | D ��} | d u r<i } | S )N�,c s s( | ] }t tt|�d ��dd��V qdS )�:N� )�tupler r �split�r[ rY r r r � <genexpr>� s �z!prepare_queues.<locals>.<genexpr>)rX r" rv �list�dictrJ r r r rQ � s
�rQ c @ sF e Zd Zddd�Zdd� Zdd � Zd
d � Zd d
� Zdd� Zdd� Z dS )�FiltererN� �?Fc s� |� _ |� _|� _|� _|� _|� _tt|�p0g �� _t |�� _
| � _ |
� _ | � _
� fdd�| pht� j
�D �� _|
pxt� � _|� _d S )Nc s g | ]}t � j|��qS r )rZ rV rw r# r r r\ s �z%Filterer.__init__.<locals>.<listcomp>)rV re �filterrf �timeout� ack_messages�setr rH rQ rB rG �foreverrP ry rk r
rh �accept)r$ rV re r} rf r~ r rH rB rG r� rP rk rh r� rW r r# r �__init__� s
� zFilterer.__init__c C sx | � | �� ��T z t| j| j| jd�D ]}q&W n$ tjyB Y n tyR Y n0 W d � n1 sh0 Y | jS )N)r~ Zignore_timeouts) �prepare_consumer�create_consumerr re r~ r� �socketr rh )r$ �_r r r �start s �
$zFilterer.startc C s. | j jd7 _| jr*| j j| jkr*t� �d S )Nr_ )rh r) rf r �r$ r3 r: r r r � update_state s zFilterer.update_statec C s |� � d S rK )ra r� r r r � ack_message# s zFilterer.ack_messagec C s | j jj| j| j| jd�S )N)rB r� )rV rR Z TaskConsumerre rk r� r# r r r r� & s
�zFilterer.create_consumerc C s� | j }| j}| j}| jr<t|| j�}t|| j�}t|| j�}|�|� |�|� | jrb|�| j� | jd ur�t| j| j �}| jr�t|| j�}|�|� | �
|� |S rK ) r} r� r� rH rI Zregister_callbackr rG r rh �declare_queues)r$ �consumerr} r� r� rG r r r r� - s$
zFilterer.prepare_consumerc C s� |j D ]t}| j r |j| j vr q| jd ur2| �|� z0||j�jdd�\}}}|r`| j j|7 _W q | jjyx Y q0 qd S )NT)Zpassive) rB rM rP rL Z
queue_declarerh r! re Zchannel_errors)r$ r� rN r� Zmcountr r r r� A s
�� zFilterer.declare_queues) Nr| FNNNFNNNN)
r r r r� r� r� r� r� r� r� r r r r r{ � s �
r{ r| Fc K s0 t | ||f||||||| |
| | |
d� |���� S )z
Filter tasks.) rf r~ r rH rB rG r� rP rk rh r� )r{ r� )rV re r} rf r~ r rH rB rG r� rP rk rh r� rW r r r r Q s" ��
�r c K s t | |ifi |��S )a� Find a task by id and move it to another queue.
Arguments:
task_id (str): Id of task to find and move.
dest: (str, kombu.Queue): Destination queue.
transform (Callable): Optional function to transform the return
value (destination) of the filter function.
**kwargs (Any): Also supports the same keyword
arguments as :func:`move`.
)r )rp rU rW r r r r f s r c s$ � fdd�}t |fdt� �i|��S )a� Move tasks by matching from a ``task_id: queue`` mapping.
Where ``queue`` is a queue to move the task to.
Example:
>>> move_by_idmap({
... '5bee6e82-f4ac-468e-bd3d-13e8600250bc': Queue('name'),
... 'ada8652d-aef3-466b-abd2-becdaf1b82b3': Queue('name'),
... '3a2b140d-7db1-41ba-ac90-c36a0ef4ab1f': Queue('name')},
... queues=['hipri'])
c s � � |jd �S )NZcorrelation_id)rA r5 rE ��mapr r �task_id_in_map� s z%move_by_idmap.<locals>.task_id_in_maprf )r �len)r� rW r� r r� r r t s r c s � fdd�}t |fi |��S )a Move tasks by matching from a ``task_name: queue`` mapping.
``queue`` is the queue to move the task to.
Example:
>>> move_by_taskmap({
... 'tasks.add': Queue('name'),
... 'tasks.mul': Queue('name'),
... })
c s � � | d �S rC )rA rE r� r r �task_name_in_map� s z)move_by_taskmap.<locals>.task_name_in_map)r )r� rW r� r r� r r � s r c K s t tjf | |d�|��� d S )N)rh r3 )�print�MOVING_PROGRESS_FMT�format)rh r3 r: rW r r r �
filter_status� s r� )ri )NNN)N)NNNNNNNN) Nr| FNNNFNNNN)-r r� � functoolsr � itertoolsr r Zkombur r Z kombu.commonr Zkombu.utils.encodingr Z
celery.appr Zcelery.utils.nodenamesr
Zcelery.utils.textr �__all__r� � Exceptionr r
r r rI r rZ r r` r r rQ r{ r r r r r� r r Zmove_direct_by_idmapZmove_direct_by_taskmapr r r r �<module> sV �
�
�
[ Z �