File: //usr/local/lib/python3.9/site-packages/celery/backends/__pycache__/elasticsearch.cpython-39.pyc
a
X>hn% � @ s� d Z ddlmZmZ ddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z z dd l
Z
W n eyr d Z
Y n0 z dd lZW n ey� d ZY n0 d
Zd ZG d d
� d
e �Zd S )z#Elasticsearch result store backend.� )�datetime�timezone�� bytes_to_str)�
_parse_url)�states)�ImproperlyConfigured� )�KeyValueStoreBackendN)�ElasticsearchBackendzVYou need to install the elasticsearch library to use the Elasticsearch result backend.c s� e Zd ZdZdZdZdZdZdZdZ dZ
dZ dZ d Z
d&� fd
d � Zd d
� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Z� fdd�Z� fd d �Zd d � Zd d!� Zd"d#� Zed$d%� �Z � Z S )'r z�Elasticsearch Backend.
Raises:
celery.exceptions.ImproperlyConfigured:
if module :pypi:`elasticsearch` is not available.
�celeryN�http� localhosti�# F�
� c s< t � j|i |�� || _| jjj}td u r2tt��d } } } } } }
} |r�t |�\}}} }
} } }
|dkrtd }| r�| �
d�} | � d�\}}
}|p�| j | _ |p�| j
| _
|p�| j| _|p�| j| _| p�| j| _|
p�| j| _| p�| j| _|d�p�| j| _|d�}|d u�r|| _|d�}|d u�r&|| _|dd�| _d | _d S )N�
elasticsearch�/Z elasticsearch_retry_on_timeoutZelasticsearch_timeoutZelasticsearch_max_retriesZ elasticsearch_save_meta_as_textT)�super�__init__�urlZapp�conf�getr r �
E_LIB_MISSINGr �strip� partition�index�doc_type�scheme�host�port�username�password�es_retry_on_timeout�
es_timeout�es_max_retries�es_save_meta_as_text�_server)�selfr �args�kwargs�_getr r r r r r r! �path�_r# r$ �� __class__� �G/usr/local/lib/python3.9/site-packages/celery/backends/elasticsearch.pyr 1 s<
�
z ElasticsearchBackend.__init__c C s2 t |tjj�r |jdv r dS t |tjj�r.dS dS )N> zN/Ai� i� i� i� � TF)�
isinstancer �
exceptionsZApiError� status_codeZTransportError)r' �excr/ r/ r0 �exception_safe_to_retryZ s
z,ElasticsearchBackend.exception_safe_to_retryc C s^ zB| � |�}z |d r&|d d W W S W n ttfy> Y n0 W n tjjyX Y n0 d S )N�found�_source�result)r* � TypeError�KeyErrorr r3 �
NotFoundError)r' �key�resr/ r/ r0 r h s
zElasticsearchBackend.getc C s2 | j r | jj| j|| j d�S | jj| j|d�S d S �N)r �idr )r r@ )r �serverr r �r' r= r/ r/ r0 r* s s ��zElasticsearchBackend._getc C s` |d� t�tj��� d d� �d�}z| j||d� W n$ tjj yZ | �
|||� Y n0 d S )Nz{}Zi����)r9 z
@timestamp)r@ �body) �formatr �nowr �utc� isoformat�_indexr r3 �
ConflictError�_update)r' r= �value�staterC r/ r/ r0 �_set_with_state� s ���
z$ElasticsearchBackend._set_with_statec C s | � ||d �S �N)rM )r' r= rK r/ r/ r0 �set� s zElasticsearchBackend.setc K sl dd� |� � D �}| jrB| jjf t|�| j| j|ddid�|��S | jjf t|�| j|ddid�|��S d S )Nc S s i | ]\}}t |�|�qS r/ r ��.0�k�vr/ r/ r0 �
<dictcomp>� � z/ElasticsearchBackend._index.<locals>.<dictcomp>Zop_type�create�r@ r r rC �params�r@ r rC rX )�itemsr rA r r )r' r@ rC r) r/ r/ r0 rH � s& �� ��zElasticsearchBackend._indexc
K s� dd� |� � D �}z0| j|d�}|�d�s@| j||fi |��W S W n* tjjyl | j||fi |�� Y S 0 z| �|d d �}W n tt fy� Y n80 |d t
j kr�ddiS |d t
j v r�|t
j
v r�ddiS |�d d
�}|�d d
�}| j�r$| jjf t|�| j| jd |i||d
�d�|��} n,| jjf t|�| jd |i||d
�d�|��} | d dk�r�tj�dt�ddt�� dt�| j| j| j��d��| S )au Update state in a conflict free manner.
If state is defined (not None), this will not update ES server if either:
* existing state is success
* existing state is a ready state and current state in not a ready state
This way, a Retry state cannot override a Success or Failure, and chord_unlock
will not retry indefinitely.
c S s i | ]\}}t |�|�qS r/ r rP r/ r/ r0 rT � rU z0ElasticsearchBackend._update.<locals>.<dictcomp>)r= r7 r8 r9 �statusZnoopZ_seq_nor Z
_primary_term�doc)Zif_primary_termZ if_seq_norW rY z(conflicting update occurred concurrentlyr1 zHTTP/1.1r N)rZ r* r rH r r3 r< Z
decode_resultr: r; r �SUCCESSZ READY_STATESZUNREADY_STATESr rA �updater r rI �elastic_transportZApiResponseMetaZ HttpHeadersZ
NodeConfigr r r )
r' r@ rC rL r) Zres_getZmeta_present_on_backendZseq_noZ prim_termr> r/ r/ r0 rJ � s\
�� ��
���z ElasticsearchBackend._updatec sp | j rt� �|�S t|t�s(t� �|�S |�d�rH| �|d �d |d< |�d�rh| �|d �d |d< |S d S )Nr9 � � traceback)r% r �encoder2 �dictr �_encode)r' �datar- r/ r0 rb � s
zElasticsearchBackend.encodec sl | j rt� �|�S t|t�s(t� �|�S |�d�rFt� �|d �|d< |�d�rdt� �|d �|d< |S d S )Nr9 ra )r% r �decoder2 rc r )r' �payloadr- r/ r0 rf � s
zElasticsearchBackend.decodec s � fdd�|D �S )Nc s g | ]}� � |��qS r/ )r )rQ r= �r' r/ r0 �
<listcomp> rU z-ElasticsearchBackend.mget.<locals>.<listcomp>r/ )r' �keysr/ rh r0 �mget s zElasticsearchBackend.mgetc C s4 | j r | jj| j|| j d� n| jj| j|d� d S r? )r rA �deleter rB r/ r/ r0 rl s zElasticsearchBackend.deletec C sL d}| j r | jr | j | jf}tj| j� d| j� d| j� �| j| j| j |d�S )z$Connect to the Elasticsearch server.Nz://�:)Zretry_on_timeout� max_retries�timeout� http_auth)
r r! r Z
Elasticsearchr r r r" r$ r# )r' rp r/ r/ r0 � _get_server
s �z ElasticsearchBackend._get_serverc C s | j d u r| �� | _ | j S rN )r& rq rh r/ r/ r0 rA s
zElasticsearchBackend.server)N) �__name__�
__module__� __qualname__�__doc__r r r r r r r! r" r# r$ r r6 r r* rM rO rH rJ rb rf rk rl rq �propertyrA �
__classcell__r/ r/ r- r0 r s4 )
B
r )ru r r Zkombu.utils.encodingr Zkombu.utils.urlr r r Zcelery.exceptionsr �baser
r � ImportErrorr_ �__all__r r r/ r/ r/ r0 �<module> s"