o
    wvXj�/  ã                   @   sà   d Z ddlmZmZ ddlZddlmZ ddlmZ ddl	m
Z
mZmZmZ ddlmZ ddlmZ dd	lmZmZ dd
lmZ ddlmZ ddlmZ dZeeƒZG dd„ deƒZdd„ Z G dd„ deƒZ!G dd„ deƒZ"dS )zHThe old AMQP result backend, deprecated and replaced by the RPC backend.é    )Úabsolute_importÚunicode_literalsN)Údeque)Ú
itemgetter)ÚConsumerÚExchangeÚProducerÚQueue)Ústates)ÚTimeoutError)Ú	monotonicÚrange)Ú
deprecated)Ú
get_loggeré   )ÚBaseBackend)ÚBacklogLimitExceededÚAMQPBackendc                   @   ó   e Zd ZdZdS )r   z'Too much state history to fast-forward.N)Ú__name__Ú
__module__Ú__qualname__Ú__doc__© r   r   úQ/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/backends/amqp.pyr      ó    r   c                 C   s:   d| d d… | dd… | dd… | dd… | dd … f S )Nz%s-%s-%s-%s-%sé   é   é   é   r   )Úsr   r   r   Úrepair_uuid   s   :r!   c                   @   r   )ÚNoCacheQueueFN)r   r   r   Úcan_cache_declarationr   r   r   r   r"   #   r   r"   c                       sD  e Zd ZdZeZeZeZeZe	Z	dZ
dZdZdddddœZ		d8‡ fdd	„	Zd9dd„Zdd„ Zdd„ Zdd„ Zdd„ Z	d:dd„Zdd„ Zddddejejfdd„Zd;dd„ZeZddedfdd „Zd<d!d"„Zd#d$„ Zddddee d%d&ƒejejfd'd(„Z!d)d*„ Z"d+d,„ Z#d-d.„ Z$d=d/d0„Z%d1d2„ Z&d>‡ fd4d5„	Z'd=d6d7„Z(‡  Z)S )?r   zbThe AMQP result backend.

    Deprecated: Please use the RPC backend or a persistent backend.
    Tr   r   r   )Úmax_retriesÚinterval_startÚinterval_stepÚinterval_maxNc           
         s�   t jddddd� tt| ƒj|fi |¤Ž | jj}	|| _|  |¡| _	| j	r'dnd| _
|p.|	j}|p3|	j}|  ||| j
¡| _|pA|	j| _|| _d S )NzThe AMQP result backendz4.0z5.0z/Please use RPC backend or a persistent backend.)ÚdeprecationÚremovalÚalternativeé   r   )r   ÚwarnÚsuperr   Ú__init__ÚappÚconfÚ_connectionÚprepare_persistentÚ
persistentÚdelivery_modeÚresult_exchangeÚresult_exchange_typeÚ_create_exchangeÚexchangeÚresult_serializerÚ
serializerÚauto_delete)
Úselfr/   Ú
connectionr8   Úexchange_typer3   r:   r;   Úkwargsr0   ©Ú	__class__r   r   r.   ?   s    þ

ÿ
zAMQPBackend.__init__Údirectr+   c                 C   s   | j |||| jdd�S )NF)ÚnameÚtyper4   Údurabler;   )r   r3   )r<   rC   rD   r4   r   r   r   r7   Q   s   üzAMQPBackend._create_exchangec                 C   s(   |   |¡}| j|| j|| j| j| jd�S )N)rC   r8   Úrouting_keyrE   r;   Úexpires)Úrkeyr	   r8   r3   r;   rG   )r<   Útask_idrC   r   r   r   Ú_create_bindingX   s   
úzAMQPBackend._create_bindingc                 C   s   d S ©Nr   )r<   Úchannelr   r   r   Úrevivec   ó   zAMQPBackend.revivec                 C   s   |  dd¡S )Nú-Ú )Úreplace©r<   rI   r   r   r   rH   f   ó   zAMQPBackend.rkeyc                 C   s&   |r|   |¡|jp
|fS |   |¡|fS rK   )rH   Úcorrelation_id)r<   rI   Úrequestr   r   r   Údestination_fori   s   zAMQPBackend.destination_forc                 K   s  |   ||¡\}}|sdS |||  ||¡||  |¡dœ}	| jj dd¡r]t|ddƒ|	d< t|ddƒ|	d< t|ddƒ|	d< t|d	dƒ|	d
< t|ddƒ|	d< t|dƒrY|jrY|j 	d¡nd|	d< | jj
jjdd��}
|
j|	| j||| jd| j|  |¡| jd�	 W d  ƒ dS 1 s‡w   Y  dS )z!Send task return value and state.N)rI   ÚstatusÚresultÚ	tracebackÚchildrenÚextendedrX   Ú	task_namerC   Úargsr?   ÚhostnameÚworkerÚretriesÚdelivery_inforF   ÚqueueT©Úblock)r8   rF   rT   r:   ÚretryÚretry_policyÚdeclarer4   )rV   Úencode_resultÚcurrent_task_childrenr/   r0   Úfind_value_for_keyÚgetattrÚhasattrra   ÚgetÚamqpÚproducer_poolÚacquireÚpublishr8   r:   rf   Úon_reply_declarer4   )r<   rI   rX   ÚstaterY   rU   r?   rF   rT   ÚpayloadÚproducerr   r   r   Ústore_resultn   s@   
ýÿþþø"ÿzAMQPBackend.store_resultc                 C   s   |   |¡gS rK   ©rJ   rR   r   r   r   rr   �   rS   zAMQPBackend.on_reply_declarec           
      K   sR   | j  |¡}	|r|	r|	d |v r|	S z
| j||||d�W S  tjy(   tdƒ‚w )NrW   )ÚtimeoutÚno_ackÚon_intervalzThe operation timed out.)Ú_cacherm   ÚconsumeÚsocketrx   r   )
r<   rI   rx   Úcachery   rz   ÚREADY_STATESÚPROPAGATE_STATESr?   Úcached_metar   r   r   Úwait_for’   s   
ÿÿzAMQPBackend.wait_foréè  c              	   C   s   | j jjdd��~\}}|  |¡|ƒ}| ¡  d  } }}t|ƒD ]"}	|j| jdd�}|s. n|jd |kr:||}}|rB| 	¡  d }q |  
|¡‚|rb|  |j¡ }
| j|< | ¡  |
W  d   ƒ S z| j| W W  d   ƒ S  ty…   tjd dœ Y W  d   ƒ S w 1 s‰w   Y  d S )NTrc   F)Úacceptry   rI   )rW   rX   )r/   ÚpoolÚacquire_channelrJ   rg   r   rm   r„   rt   Úackr   Úmeta_from_decodedr{   ÚrequeueÚKeyErrorr
   ÚPENDING)r<   rI   Úbacklog_limitÚ_rL   ÚbindingÚprevÚlatestÚaccÚirt   r   r   r   Úget_task_meta¡   s@   ÿ
€
ÿç
ãàâzAMQPBackend.get_task_metac           	         s”   |p|j }i ‰ ‡ ‡fdd„}|g|jd d …< |ƒ }	 |r'|ƒ | |kr't ¡ ‚z|dd� W n
 tjy8   Y nw |r>|ƒ  ˆ rAnqˆj ˆ ¡ ˆ S )Nc                    s(   | d t jv rˆ | ¡ˆ | d < d S d S )NrW   rI   )r
   r   rˆ   )ÚmetaÚmessage©Úresultsr<   r   r   ÚcallbackË   s   ÿz*AMQPBackend.drain_events.<locals>.callbackr   ©rx   )Údrain_eventsÚ	callbacksr}   rx   r{   Úupdate)	r<   r=   Úconsumerrx   rz   ÚnowÚwaitr˜   Ú
time_startr   r–   r   rš   Æ   s(   
ÿõzAMQPBackend.drain_eventsc           
   
   C   s¶   | j }| jjjdd��F\}}|  |¡}| j|||| jd��$}		 z|||	||ƒ| W W  d   ƒ W  d   ƒ S  ty@   Y qw 1 sDw   Y  W d   ƒ d S 1 sTw   Y  d S )NTrc   )ry   r„   )rš   r/   r…   r†   rJ   r   r„   rŠ   )
r<   rI   rx   ry   rz   rŸ   ÚconnrL   rŽ   r�   r   r   r   r|   á   s*   
ÿÿÿüþÿú"þzAMQPBackend.consumec                    s   ‡ fdd„|D ƒS )Nc                    s   g | ]}ˆ   |¡‘qS r   rw   )Ú.0rI   ©r<   r   r   Ú
<listcomp>ï   s    z.AMQPBackend._many_bindings.<locals>.<listcomp>r   )r<   Úidsr   r£   r   Ú_many_bindingsî   s   zAMQPBackend._many_bindingsrW   rI   c
              
   +   s‚  � | j jjdd��®\}}tˆƒ}tƒ }|j}|D ]"}z| j| }W n	 ty*   Y qw |d ˆ v r:||fV  ||ƒ q| |¡ tƒ }|j	‰| jj
‰| j‰‡ ‡‡‡‡‡‡fdd„}|  ˆ¡}| j|||| j|d��6 |j}|j}|r˜||d� |r‘|ƒ }|d }| |¡ ˆ||ƒ ||fV  |sy|r–|ƒ  |srW d   ƒ n1 s¢w   Y  W d   ƒ d S W d   ƒ d S 1 sºw   Y  d S )	NTrc   rW   c                    sR   ˆ|   ¡ ƒ}ˆd urˆ|ƒ ˆ|ƒ\}}|ˆ v r'|ˆv r ˆ|ƒnˆ||ƒ d S d S rK   )Údecode)r•   Úbodyrs   Úuid©r   Údecode_resultÚ	getfieldsÚ
on_messageÚ
push_cacheÚpush_resultÚtask_idsr   r   Ú_on_message	  s   ÿ€ þz)AMQPBackend.get_many.<locals>._on_message)r­   r„   ry   r™   rI   )r/   r…   r†   ÚsetÚaddr{   rŠ   Údifference_updater   ÚappendÚ__setitem__rˆ   r¦   r   r„   rš   ÚpopleftÚdiscard)r<   r°   rx   ry   r­   rz   rž   r¬   r   r€   r?   r¡   rL   r¥   Ú
cached_idsÚmark_cachedrI   Úcachedr—   r±   ÚbindingsrŸ   r·   rs   r   rª   r   Úget_manyñ   sX   €ÿ
€

	
ÿ



û÷€üã"ãzAMQPBackend.get_manyc                 C   ó   t dƒ‚)Nz4reload_task_result is not supported by this backend.©ÚNotImplementedErrorrR   r   r   r   Úreload_task_result"  ó   ÿzAMQPBackend.reload_task_resultc                 C   r¾   )z<Reload group result, even if it has been previously fetched.z5reload_group_result is not supported by this backend.r¿   rR   r   r   r   Úreload_group_result&  s   ÿzAMQPBackend.reload_group_resultc                 C   r¾   )Nz,save_group is not supported by this backend.r¿   )r<   Úgroup_idrX   r   r   r   Ú
save_group+  rÂ   zAMQPBackend.save_groupc                 C   r¾   )Nz/restore_group is not supported by this backend.r¿   )r<   rÄ   r~   r   r   r   Úrestore_group/  rÂ   zAMQPBackend.restore_groupc                 C   r¾   )Nz.delete_group is not supported by this backend.r¿   )r<   rÄ   r   r   r   Údelete_group3  rÂ   zAMQPBackend.delete_groupr   c              	      sH   |r|ni }|j | j| jj| jj| j| j| j| jd� t	t
| ƒ ||¡S )N)r=   r8   r>   r3   r:   r;   rG   )rœ   r1   r8   rC   rD   r3   r:   r;   rG   r-   r   Ú
__reduce__)r<   r]   r?   r@   r   r   rÈ   7  s   ù	zAMQPBackend.__reduce__c                 C   s   dS )Nzamqp://r   )r<   Úinclude_passwordr   r   r   Úas_uriD  rN   zAMQPBackend.as_uri)NNNNNT)rB   r+   )NN)rƒ   )NTN)T)r   N)*r   r   r   r   r   r"   r	   r   r   r   r3   Úsupports_autoexpireÚsupports_native_joinrf   r.   r7   rJ   rM   rH   rV   rv   rr   r
   r   r€   r‚   r“   Úpollr   rš   r|   r¦   r   r½   rÁ   rÃ   rÅ   rÆ   rÇ   rÈ   rÊ   Ú__classcell__r   r   r@   r   r   '   sd    üÿ

ÿ!
ý
#
ÿ


ü1
r   )#r   Ú
__future__r   r   r}   Úcollectionsr   Úoperatorr   Úkombur   r   r   r	   Úceleryr
   Úcelery.exceptionsr   Úcelery.fiver   r   Úcelery.utilsr   Úcelery.utils.logr   Úbaser   Ú__all__r   ÚloggerÚ	Exceptionr   r!   r"   r   r   r   r   r   Ú<module>   s$   