o wvXj[ @sdZddlmZmZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZdd lmZmZdd lmZd d lmZzddlZWn eyUdZYnwed ZdZdZGdddejZGdddejZdS)z}Etcd Transport. It uses Etcd as a store to transport messages in Queues It uses python-etcd for talking to Etcd's HTTP API )absolute_importunicode_literalsN) defaultdict)contextmanager) ChannelError)Empty) get_logger)loadsdumps)cached_property)virtualzkombu.transport.etcdiK localhostcseZdZdZdZdZdZdZdZfddZ dd Z e d d Z d d Z ddZddZddZdddZddZddZeddZZS)Channelz+Etcd Channel class which talks to the Etcd.kombuN csztdurtdtt|j|i||jjjp|jj}|jjj p"t }t d|||j tt|_tj|t|d|_dS)NMissing python-etcd libraryzHost: %s Port: %s Timeout: %shostport)etcd ImportErrorsuperr__init__ connectionclientr default_porthostname DEFAULT_HOSTloggerdebugtimeoutrdictqueuesClientint)selfargskwargsrr __class__Q/var/www/html/myproject/venv/lib/python3.10/site-packages/kombu/transport/etcd.pyr+s zChannel.__init__cCsd|j|S)zCreate and return the `queue` with the proper prefix. Arguments: queue (str): The name of the queue. z{0}/{1})formatprefix)r'queuer,r,r- _key_prefix:szChannel._key_prefixc cs~t|j|}|j|_td|j|j d|j dzdVWtd|j| dStd|j| w)agTry to acquire a lock on the Queue. It does so by creating a object called 'lock' which is locked by the current session.. This way other nodes are not able to write to the lock object which means that they have to wait before the lock is released. Arguments: queue (str): The name of the queue. zAcquiring lock {0}T)blockinglock_ttlNzReleasing lock {0}) rLockr lock_value_uuidr r!r.nameacquirer3release)r'r0lockr,r,r- _queue_lockBs   zChannel._queue_lockc Ks||j|<||9z|jj||dddWWdStjyBtd ||jj ||dYWdSw1sFwYdS)zCreate a new `queue` if the `queue` doesn't already exist. Arguments: queue (str): The name of the queue. TN)keydirvaluezQueue "{0}" already existsr<) r$r;rwriter1r EtcdNotFiler r!r.read)r'r0_r,r,r- _new_queueYs   zChannel._new_queuecKs0z |j||WdStjyYdSw)zVerify that queue exists. Returns: bool: Should return :const:`True` if the queue exists or :const:`False` otherwise. TF)rrBr1rEtcdKeyNotFound)r'r0r)r,r,r- _has_queuehs zChannel._has_queuecOs|j|d||dS)z^Delete a `queue`. Arguments: queue (str): The name of the queue. N)r$pop_purge)r'r0r(rCr,r,r-_deleteuszChannel._deletecKs^|| ||}|jj|t|ddstd|WddS1s(wYdS)zPut `message` onto `queue`. This simply writes a key to the Etcd store Arguments: queue (str): The name of the queue. payload (dict): Message data which will be dumped to etcd. T)r<r>appendzCannot add key {0!r} to etcdN)r;r1rr@r rr.)r'r0payloadrCr<r,r,r-_put~s "z Channel._putc Cs||m||}td||jz;|jj|d|j|jd}|dur't|j d}td |dt |d}|jj |dd |WWdSt ttjfyq}ztd t||WYd}~td}~ww1suwYdS) aIGet the first available message from the queue. Before it does so it acquires a lock on the store so only one node reads at the same time. This is for read consistency Arguments: queue (str): The name of the queue. timeout (int): Optional seconds to wait for a response. zFetching key %s with index %sT)r< recursiveindexr"NzRemoving key {0}r<r>r?z_get failed: {0}:{1})r;r1r r!rNrrBr"r _childrenr.r delete TypeError IndexErrorr EtcdExceptiontype)r'r0r"r<resultitem msg_contenterrorr,r,r-_gets,    z Channel._getcCsX||||}td||jj|ddWdS1s%wYdS)zrRemove all `message`s from a `queue`. Arguments: queue (str): The name of the queue. zPurging queue at key {0}T)r<rMN)r;r1r r!r.rrQ)r'r0r<r,r,r-rHs  $zChannel._purgec Cs||=d}z||}td||j|jj|d|jd}t|j}Wn t y/Ynwtd|||j|WdS1sEwYdS)zlReturn the size of the `queue`. Arguments: queue (str): The name of the queue. rz)Fetching key recursively %s with index %sT)r<rMrNz$Found %s keys under %s with index %sN) r;r1r r!rNrrBlenrPrR)r'r0sizer<rVr,r,r-_sizes(   $z Channel._sizecCsdttS)Nz{0}.{1})r.socket gethostnameosgetpid)r'r,r,r-r5szChannel.lock_value)N)__name__ __module__ __qualname____doc__r/rNr" session_ttlr3rr1rr;rDrFrIrLrZrHr]r r5 __classcell__r,r,r*r-r"s(    ! rcsZeZdZdZeZeZdZdZdZ e j j j edgdZ fddZd d Zd d ZZS) Transportz!Etcd storage Transport for Kombu.r python-etcddirect) exchange_typecsNtdurtdtt|j|i|tjjtjf|_tjjtjf|_dS)z(Create a new instance of etcd.Transport.Nr) rrrrhrr connection_errorsrTchannel_errors)r'r(r)r*r,r-rszTransport.__init__cCsV|jjp|j}|jjp t}td||z tj|t |dWdSt y*YdSw)zVerify the connection works.zVerify Etcd connection to %s:%srTF) rrrrrr r!rr%r& ValueError)r'rrrr,r,r-verify_connections  zTransport.verify_connectionc Csbzddl}|jjD]}|dr|ddWSq WdSttfy0tdYdSw)zReturn the version of the etcd library. .. note:: python-etcd has no __version__. This is a workaround. rNriz==r z'Unable to find the python-etcd version.Unknown) pip.commands.freezecommandsfreeze startswithsplitrrSr warning)r'pipxr,r,r-driver_versions  zTransport.driver_version)rbrcrdrer DEFAULT_PORTr driver_type driver_namepolling_intervalr rh implementsextend frozensetrrprzrgr,r,r*r-rhs rh)re __future__rrr`r^ collectionsr contextlibrkombu.exceptionsr kombu.fiver kombu.logrkombu.utils.jsonr r kombu.utils.objectsr r rrr r{rrrhr,r,r,r-s.         6