File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/mongodb.cpython-39.pyc
a
X>h�= � @ s� d Z ddlmZ ddlZddlmZ ddlZddlmZmZm Z ddl
m Z ddl m
Z
ddlmZ dd lmZ dd
lmZmZ dd lmZ dd lmZ d
dlmZ d
dlm Z dZ dZ G dd� d�Z G dd� dej �Z G dd� dej!�Z!dS )a MongoDB transport module for kombu.
Features
========
* Type: Virtual
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: Yes
* Supports Priority: Yes
* Supports TTL: Yes
Connection String
=================
*Unreviewed*
Transport Options
=================
* ``connect_timeout``,
* ``ssl``,
* ``ttl``,
* ``capped_queue_size``,
* ``default_hostname``,
* ``default_port``,
* ``default_database``,
* ``messages_collection``,
* ``routing_collection``,
* ``broadcast_collection``,
* ``queues_collection``,
* ``calc_queue_size``,
� )� annotationsN)�Empty)� MongoClient�errors�
uri_parser)�
CursorType)�VersionMismatch)�_detect_environment)� bytes_to_str)�dumps�loads)�cached_property)�maybe_sanitize_url� )�virtual��to_rabbitmq_queue_argumentsz3Kombu requires MongoDB version 1.3+ (server is {0})zKKombu requires MongoDB version 2.2+ (server is {0}) for TTL indexes supportc @ sF e Zd ZdZdd� Zdd� Zdd� Zdd d
�Zd d � Zd
d� Z e Z
dS )�BroadcastCursorz Cursor for broadcast queues.c C s || _ d| _| jdd� d S )Nr F)�rewind)�_cursor�_offset�purge)�self�cursor� r �A/usr/local/lib/python3.9/site-packages/kombu/transport/mongodb.py�__init__C s zBroadcastCursor.__init__c C s | j j�i �| j S �N)r �
collection�count_documentsr �r r r r �get_sizeH s zBroadcastCursor.get_sizec C s | j �� d S r )r �closer r r r r" K s zBroadcastCursor.closeTc C s2 |r| j �� | j j�i �| _| j �| j�| _ d S r )r r r r r �skip)r r r r r r N s
zBroadcastCursor.purgec C s | S r r r r r r �__iter__V s zBroadcastCursor.__iter__c
C sp zt | j�}W q^ tjjyX } z.dt|�v rB| �� W Y d }~q � W Y d }~q d }~0 0 q^q | jd7 _|S )Nznot valid at serverr )�nextr �pymongor �OperationFailure�strr r )r �msg�excr r r �__next__Y s zBroadcastCursor.__next__N)T) �__name__�
__module__� __qualname__�__doc__r r! r" r r$ r+ r% r r r r r @ s
r c s\ e Zd ZdZdZi ZdZdZdZdZ dZ
dZ dZ dZ
d Zd
Zd Zd Zejjd
Z� fdd�Zdd� Zdd� Z� fdd�Zdd� Zdd� Zdd� Zd d � Z d d � Z � fd d!�Z dGd#d$�Z d%d&� Z d'd(� Z!dHd)d*�Z"d+d,� Z#d-d.� Z$d/d0� Z%e&d1d2� �Z'e&d3d4� �Z(e&d5d6� �Z)e&d7d8� �Z*e&d9d:� �Z+d;d<� Z,d=d>� Z-d?d@� Z.dAdB� Z/dCdD� Z0dEdF� Z1� Z2S )I�ChannelzMongoDB Channel.TFNi�� z 127.0.0.1i�i Z
kombu_default�messageszmessages.routingzmessages.broadcastzmessages.queues) �connect_timeout�ssl�ttl�capped_queue_size�default_hostname� default_port�default_database�messages_collection�routing_collection�broadcast_collection�queues_collection�calc_queue_sizec s" t � j|i |�� i | _| j d S r )�superr �_broadcast_cursors�client)r Zvargs�kwargs�� __class__r r r � s zChannel.__init__c K s4 | j r0| jjd|id||| �|d�d�idd� d S )N�_id�$set� x-expires)rD �options� expire_atT�Zupsert)r4 �queues�
update_one�_get_queue_expire)r �queuerA r r r �
_new_queue� s ��� �zChannel._new_queuec C s� || j v r6zt| �|��}W qR ty2 d }Y qR0 n | jjd|idtjfgd�}| jrb| � |� |d u rpt
� �t t |d ��S )NrM �priority)�sort�payload)
�_fanout_queuesr% �_get_broadcast_cursor�
StopIterationr1 Zfind_one_and_deleter& Z ASCENDINGr4 �_update_queues_expirer r r
)r rM r) r r r �_get� s
�
z Channel._getc s: | j st� �|�S || jv r*| �|��� S | j�d|i�S �NrM )r= r> �_sizerR rS r! r1 r �r rM rB r r rX � s
z
Channel._sizec K sr t |�|| j|dd�d�}| jrb| �|d�|d< | �|�}|d urb|d d u sZ||d k rb||d< | j�|� d S )NT)�reverse)rQ rM rO z
x-message-ttlrH )r Z_get_message_priorityr4 rL �_get_message_expirer1 �
insert_one)r rM �messagerA �dataZ
msg_expirer r r �_put� s �
�
�z Channel._putc K s | j �t|�|d�� d S )N)rQ rM )� broadcastr\ r )r �exchanger] � routing_keyrA r r r � _put_fanout� s �zChannel._put_fanoutc C s8 | � |�}|| jv r$| �|��� n| j�d|i� |S rW )rX rR rS r r1 � delete_many)r rM �sizer r r �_purge� s
zChannel._purgec C s: t | jj| d �}| j�d|i�}|t dd� |D ��B S )N�tablera c s s$ | ] }|d |d |d fV qdS )rb �patternrM Nr )�.0�rr r r � <genexpr>� s �z$Channel.get_table.<locals>.<genexpr>)� frozenset�stateZ exchanges�routing�find)r ra Z localRoutesZ brokerRoutesr r r � get_table� s �
�zChannel.get_tablec C sp | � |�jdkr*| �||||� || j|< ||||d�}|�� }| jrV| �|d�|d< | jj|d|idd� d S )N�fanout)ra rM rb rh rF rH rE TrI ) Ztypeof�type�_create_broadcast_cursorrR �copyr4 rL rn rK )r ra rb rh rM �lookupr^ r r r � _queue_bind� s �
�zChannel._queue_bindc s� | j �d|i� | jr&| j�d|i� t� j|fi |�� || jv r|z| j� |�}W n t
yf Y n0 |� � | j� |� d S )NrM rD ) rn rd r4 rJ Z
delete_oner> � queue_deleterR r? �pop�KeyErrorr" )r rM rA r rB r r rw s
zChannel.queue_delete�
mongodb://c C s0 | j j}|j}|�d�r$d}d| }|�|�s6|| }|t|�d � sP|| j7 }|jr�d|vr�|�d�\}}|j}|jr�|d|j 7 }|d | d | }|j r�|j n| j
}t j ||dd�}|d p�|j
} | d
v r�| j} d | j| jr�t| jd �nd d
�}
|
�|d � | �|
�}
d|
v �r&|
�d� || |
fS )Nzsrv://zmongodb+srv://zmongodb+�@z://�:F)�validate�database)�/NTi� )�auto_start_requestr3 ZconnectTimeoutMSrG Ztlsr3 )�
connectionr@ �hostname�
startswith�lenr6 Zuserid�split�password�portr7 r � parse_uriZ virtual_hostr8 r3 r2 �int�update�_prepare_client_optionsrx ) r �schemer@ r� �head�tail� credentialsr� �parsed�dbnamerG r r r �
_parse_uri s>
��
zChannel._parse_uric C sB t jdkr>|�dd � t|�d�t�r>t jj}||d |d< |S )N�� r� Zreadpreference)r& �
version_tuplerx �
isinstance�getr� Zread_preferencesZ
_MONGOS_MODES)r rG �modesr r r r� K s
z Channel._prepare_client_optionsc K s t |fi |��S r r )r � argumentsrA r r r �prepare_queue_argumentsS s z Channel.prepare_queue_argumentsc C s� | j |d�\}}}||d< t� }|dkr>ddlm} |�� n|dkrXddlm} |� tf i |��}|| } |�� d }
|
� d �d }
t
t t |
� d
���} | d k r�t
t�|
���n | jr�| d k r�t
t�|
���| S )
N)r� �host�geventr )�monkey�eventlet)� monkey_patch�version�-�.)r r� )� r� )r� r r� r� Z patch_allr� r� r Z server_infor� �tuple�mapr� r �E_SERVER_VERSION�formatr4 �E_NO_TTL_INDEXES) r r� r� r� �conf�envr� r� Z mongoconnr~ � version_strr� r r r �_openV s&
z
Channel._openc C s* | j |�� v rdS |j| j | jdd� dS )z0Create capped collection for broadcast messages.NT)re Zcapped)r; Zlist_collection_namesZcreate_collectionr5 �r r~ r r r �_create_broadcastq s �zChannel._create_broadcastc C s� || j }|jg d�dd� || j �dg� || j }|�ddg� | jr�|jdgdd� |jdgdd� || j jdgdd� d S )
z Ensure indexes on collections.)�rM r )rO r )rD r T)Z
backgroundr� )ra r )rH r r )ZexpireAfterSecondsN)r9 Z create_indexr; r: r4 r< )r r~ r1 rn r r r �_ensure_indexesz s
�
�zChannel._ensure_indexesc C s | � � }| �|� | �|� |S )z Actually creates connection.)r� r� r� r� r r r �_create_client� s
zChannel._create_clientc C s | � � S r )r� r r r r r@ � s zChannel.clientc C s | j | j S r )r@ r9 r r r r r1 � s zChannel.messagesc C s | j | j S r )r@ r: r r r r rn � s zChannel.routingc C s | j | j S r )r@ r; r r r r r` � s zChannel.broadcastc C s | j | j S r )r@ r< r r r r rJ � s zChannel.queuesc C s: z | j | W S ty4 | �| j| d d |� Y S 0 d S r )r? ry rs rR rY r r r rS � s �z Channel._get_broadcast_cursorc C sR t jdkr d|itjd�}nd|idd�}| jjf i |��}t|� }| j|< |S )Nr� rM )�filterZ cursor_typeT)�queryZtailable)r& r� r ZTAILABLEr` ro r r? )r ra rb rh rM r� r �retr r r rs � s
��z Channel._create_broadcast_cursorc C s6 |� di �� d�}|d ur2| �� tjt|�d� S d S )NZ
propertiesZ
expiration�Z milliseconds)r� �get_now�datetime� timedeltar� )r r] �valuer r r r[ � s zChannel._get_message_expirec C sn t |t�r,| j�d|i�}|s"dS |d }n|}z|d | }W n ttfyX Y dS 0 | �� tj|d� S )z�Get expiration header named `argument` of queue definition.
Note:
----
`queue` must be either queue name or options itself.
rD NrG r� r� ) r� r( rJ Zfind_onery � TypeErrorr� r� r� )r rM �argument�docr^ r� r r r rL � s
zChannel._get_queue_expirec C sL | � |d�}|sdS | j�d|idd|ii� | j�d|idd|ii� dS )z,Update expiration field on queues documents.rF NrM rE rH rD )rL rn Z update_manyrJ )r rM rH r r r rU � s ��z Channel._update_queues_expirec C s
t j �� S )zReturn current time in UTC.)r� �utcnowr r r r r� � s zChannel.get_now)rz )rz )3r, r- r. r/ Zsupports_fanoutrR r3 r4 r2 r5 r= r6 r7 r8 r9 r: r; r< r r0 Zfrom_transport_optionsr rN rV rX r_ rc rf rp rv rw r� r� r� r� r� r� r� r
r@ r1 rn r` rJ rS rs r[ rL rU r� �
__classcell__r r rB r r0 o sb
2
r0 c @ s� e Zd ZdZeZdZdZejZej j
e j f Z
ej j
e j e jf Z
dZdZej jjeg d��d�Zdd � Zdd d d
�dd�ZdS )� TransportzMongoDB Transport.Tr Zmongodbr& )�directZtopicrq )Z
exchange_typec C s t jS r )r& r� r r r r �driver_version s zTransport.driver_versionF�**r( )�uri�returnc C sB |sdS |r|S d|vr t |�S |�dd�\}}d�t |�|g�S )Nrz �,r )r r� �join)r r� Zinclude_password�maskZuri1� remainderr r r �as_uri s zTransport.as_uriN)Fr� )r, r- r. r/ r0 Z
can_parse_urlZpolling_intervalr7 r r� Zconnection_errorsr ZConnectionFailureZchannel_errorsr' Z driver_typeZ driver_nameZ
implements�extendrl r� r� r r r r r� � s& ���
�r� )"r/ �
__future__r r� rM r r& r r r Zpymongo.cursorr Zkombu.exceptionsr Zkombu.utils.compatr Zkombu.utils.encodingr
Zkombu.utils.jsonr r Zkombu.utils.objectsr
Zkombu.utils.urlr � r �baser r� r� r r0 r� r r r r �<module> s, /