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/backends/__pycache__/ |
a
��Sh-/ � @ s� d Z ddlZddlZddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z m
Z
dd lmZ dd
lmZmZ d Zd ZG d
d� de�Zdd� ZG dd� de�ZG dd� deje�ZdS )zqThe ``RPC`` result backend for AMQP brokers.
RPC-style result backend, using reply-to and one queue per client.
� N)�
maybe_declare)�register_after_fork)�cached_property)�states)� current_task�task_join_will_block� )�base)�AsyncBackendMixin�BaseResultConsumer)�BacklogLimitExceeded�
RPCBackendz�
The "rpc" result backend does not support chords!
Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.
Result backends that supports chords: Redis, Database, Memcached, and more.
c @ s e Zd ZdZdS )r z'Too much state history to fast-forward.N)�__name__�
__module__� __qualname__�__doc__� r r �=/usr/local/lib/python3.9/site-packages/celery/backends/rpc.pyr s r c C s | � � d S �N)� _after_fork)�backendr r r � _on_after_fork_cleanup_backend" s r c s^ e Zd ZejZdZdZ� fdd�Zddd�Zddd�Z d d
� Z
d d � Z d
d� Z dd� Z
� ZS )�ResultConsumerNc s t � j|i |�� | jj| _d S r )�super�__init__r �_create_binding��self�args�kwargs�� __class__r r r , s zResultConsumer.__init__Tc K sF | j �� | _| �|�}| j| jj|g| jg|| jd�| _| j� � d S )N)� callbacks�no_ack�accept)
�app�
connection� _connectionr �ConsumerZdefault_channelZon_state_changer$ � _consumer�consume)r Zinitial_task_idr# r Z
initial_queuer r r �start0 s
�zResultConsumer.startc C s& | j r| j j|d�S |r"t�|� d S )N)�timeout)r' � drain_events�time�sleep)r r, r r r r- 9 s zResultConsumer.drain_eventsc C s* z| j �� W | j�� n | j�� 0 d S r )r) �cancelr' �close�r r r r �stop? s zResultConsumer.stopc C s$ d | _ | jd ur | j�� d | _d S r )r) r' Zcollectr2 r r r �
on_after_forkE s
z ResultConsumer.on_after_forkc C sD | j d u r| �|�S | �|�}| j �|�s@| j �|� | j �� d S r )r) r+ r Zconsuming_fromZ add_queuer* )r �task_id�queuer r r � consume_fromK s
zResultConsumer.consume_fromc C s | j r| j �| �|�j� d S r )r) Zcancel_by_queuer �name�r r5 r r r �
cancel_forS s zResultConsumer.cancel_for)T)N)r r r �kombur( r' r) r r+ r- r3 r4 r7 r: �
__classcell__r r r r r & s
r c sZ e Zd ZdZejZejZeZeZdZ dZ
dZ ddddd�Z G dd � d ej
�Z
G d
d � d ej�ZdE� fd
d� Zdd� ZdFdd�Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � ZdGd!d"�ZdHd#d$�Zd%d&� Zd'd(� ZdId*d+�Z e Z d,d-� Z dJd.d/�Z d0d1� Z d2d3� Z!d4d5� Z"d6d7� Z#d8d9� Z$dKd:d;�Z%d<d=� Z&dL� fd?d@� Z'e(dAdB� �Z)e*dCdD� �Z+� Z,S )Mr
z&Base class for the RPC result backend.FT� r r )� max_retriesZinterval_startZ
interval_stepZ interval_maxc @ s e Zd ZdZdZdS )zRPCBackend.Consumerz4Consumer that requires manual declaration of queues.FN)r r r r Z auto_declarer r r r r( m s r( c @ s e Zd ZdZdZdS )zRPCBackend.Queuez$Queue that never caches declaration.FN)r r r r Zcan_cache_declarationr r r r �Queuer s r? Nc
s� t � j|fi |�� | jj} || _i | _| �|�| _| jr>dnd| _|pL| j }|pV| j
}| � ||| j�| _ |pr| j
| _|| _| �| | j| j| j| j�| _td ur�t| t� d S )N� r )r r r% �confr' � _out_of_bandZprepare_persistent�
persistent�
delivery_modeZresult_exchangeZresult_exchange_type�_create_exchange�exchangeZresult_serializer�
serializer� auto_deleter r$ �_pending_resultsZ_pending_messages�result_consumerr r )
r r% r& rF �
exchange_typerC rG rH r rA r r r r w s&
�
�zRPCBackend.__init__c C s | j �� | j�� d S r )rI �clearrJ r r2 r r r r � s
zRPCBackend._after_fork�directr@ c C s
| � d �S r )�Exchange)r r8 �typerD r r r rE � s zRPCBackend._create_exchangec C s | j S )z$Create new binding for task with id.)�bindingr9 r r r r � s zRPCBackend._create_bindingc C s t t�� ��d S r )�NotImplementedError�E_NO_CHORD_SUPPORT�stripr2 r r r �ensure_chords_allowed� s z RPCBackend.ensure_chords_allowedc C s t � st| �|j�dd� d S )NT)�retry)r r rP �channel)r �producerr5 r r r � on_task_call� s zRPCBackend.on_task_callc C s@ z|p
t j}W n ty. td|����Y n0 |j|jp<|fS )z�Get the destination for result by task id.
Returns:
Tuple[str, str]: tuple of ``(reply_to, correlation_id)``.
z%RPC backend missing task request for )r �request�AttributeError� RuntimeErrorZreply_to�correlation_id)r r5 rY r r r �destination_for� s �
zRPCBackend.destination_forc C s d S r r r9 r r r �on_reply_declare� s zRPCBackend.on_reply_declarec C s d S r r )r �resultr r r �on_result_fulfilled� s z RPCBackend.on_result_fulfilledc C s dS )Nzrpc://r )r Zinclude_passwordr r r �as_uri� s zRPCBackend.as_uric
K s� | � ||�\}}|sdS | jjjjdd��H} | j| �|||||�| j||| jd| j | �
|�| j d� W d � n1 sx0 Y |S )z!Send task return value and state.NT��block)rF � routing_keyr\ rG rU � retry_policy�declarerD ) r] r% ZamqpZ
producer_pool�acquire�publish�
_to_resultrF rG re r^ rD )
r r5 r_ �state� tracebackrY r rd r\ rW r r r � store_result� s �$
zRPCBackend.store_resultc C s ||| � ||�|| �|�d�S )N)r5 �statusr_ rk �children)Z
encode_resultZcurrent_task_children)r r5 rj r_ rk rY r r r ri � s
�zRPCBackend._to_resultc C s | j r| j �|� || j|< d S r )rJ �on_out_of_band_resultrB )r r5 �messager r r ro � s z RPCBackend.on_out_of_band_result�� c
C s� | j �|d �}|r | �||�S i }d }| �|| j|�D ]2}| �|�}|�|�| }||< |r6|�� d }q6|�|d �}|�� D ]\}} | � || � q~|r�|�
� | �||�S z | j | W S t y� t
jd d� Y S 0 d S )N)rm r_ )rB �pop�_set_cache_by_message�_slurp_from_queuer$ �_get_message_task_id�getZack�itemsro Zrequeue�_cache�KeyErrorr �PENDING)
r r5 Z
backlog_limitZbufferedZ latest_by_id�prev�acc�tidZlatest�msgr r r �
get_task_meta� s*
zRPCBackend.get_task_metac C s | � |j� }| j|< |S r )Zmeta_from_decoded�payloadrx )r r5 rp r� r r r rs s �z RPCBackend._set_cache_by_messagec c s� | j jjdd��^\}}| �|�|�}|�� t|�D ] }|j||d�}|sN q`|V q4| �|��W d � n1 st0 Y d S )NTrb )r$ r# )r% �poolZacquire_channelr rf �rangerv r ) r r5 r$ �limitr# �_rV rP r~ r r r rt s z RPCBackend._slurp_from_queuec C s2 z |j d W S ttfy, |jd Y S 0 d S )Nr\ r5 )Z
propertiesrZ ry r� )r rp r r r ru s z RPCBackend._get_message_task_idc C s d S r r )r rV r r r �revive% s zRPCBackend.revivec C s t d��d S )Nz4reload_task_result is not supported by this backend.�rQ r9 r r r �reload_task_result( s �z RPCBackend.reload_task_resultc C s t d��dS )z<Reload group result, even if it has been previously fetched.z5reload_group_result is not supported by this backend.Nr� r9 r r r �reload_group_result, s �z RPCBackend.reload_group_resultc C s t d��d S )Nz,save_group is not supported by this backend.r� )r �group_idr_ r r r �
save_group1 s �zRPCBackend.save_groupc C s t d��d S )Nz/restore_group is not supported by this backend.r� )r r� �cacher r r �
restore_group5 s �zRPCBackend.restore_groupc C s t d��d S )Nz.delete_group is not supported by this backend.r� )r r� r r r � delete_group9 s �zRPCBackend.delete_groupr c
s@ |si n|}t � �|t|| j| jj| jj| j| j| j | j
d��S )N)r& rF rK rC rG rH �expires) r �
__reduce__�dictr' rF r8 rO rC rG rH r� r r r r r� = s
�zRPCBackend.__reduce__c C s | j | j| j| jdd| jd�S )NFT)ZdurablerH r� )r? �oidrF r� r2 r r r rP J s �zRPCBackend.bindingc C s | j jS r )r% Z
thread_oidr2 r r r r� S s zRPCBackend.oid)NNNNNT)rM r@ )T)NN)rq )rq F)T)r N)-r r r r r; rN ZProducerr r rC Zsupports_autoexpireZsupports_native_joinre r( r? r r rE r rT rX r] r^ r` ra rl ri ro r �pollrs rt ru r� r� r� r� r� r� r� �propertyrP r r� r< r r r r r
X s` � �
�
�
r
)r r. r; Z kombu.commonr Zkombu.utils.compatr Zkombu.utils.objectsr Zceleryr Z
celery._stater r � r Z asynchronousr
r �__all__rR � Exceptionr r r ZBackendr
r r r r �<module> s
2