o wOiD@sdZddlZzddlmZWneyddlmZYnwddlZddlZddlm Z ddlm Z ddl Z ddl Z ddl ZddlmZddlZzddlmZWney_ddlmZYnweeZGdd d eZdS) zp .. module:: python-etcd :synopsis: A python etcd client. .. moduleauthor:: Jose Plana N) HTTPException) HTTPError)ReadTimeoutErrorwraps)urlparsec@s eZdZdZdZdZdZdZedZ edZ edZ d Z      dlddZ ddZddZddZeddZeddZedd Zed!d"Zed#d$Zed%d&Zed'd(Zed)d*Zed+d,Zed-d.Zed/d0Zed1d2Zed3d4Zdmd6d7Zed8d9Z ed:d;Z!edd?Z#d@dAZ$dndBdCZ%dDdEZ&dFdGZ'dHdIZ(dodJdKZ)dodLdMZ*dpdNdOZ+dpdPdQZdRdSZ,dqdTdUZ-dodVdWZ.dXdYZ/edZd[Z0d\d]Z1dpd^d_Z2d`daZ3e3dodbdcZ4e3dodddeZ5dfdgZ6dhdiZ7djdkZ8d S)rClientzB Client for etcd, the distributed log service using raft. GETPUTPOSTDELETE) prevValue prevIndex prevExistrefresh) recursivewait waitIndexsortedquorum)r rN 127.0.0.1/v2<ThttpF /_locksc s|dur%z|}Wnty$}z td||WYd}~nd}~ww|_ddt|ts=g_j||_n| sItdt dfdd|D_j d _|_ |_ |_|_| _| _|_d |i}jd kr~j|d <|rt|tr|d |d <|d |d<n||d <| r| |d<tj|d<d_d_| r| r| _| _n| rtdn| rtdtjdddi|_tdjjrjsttjtjB_jjvrj jtdjd_!_"dS)a Initialize the client. Args: host (mixed): If a string, IP to connect to. If a tuple ((host, port), (host, port), ...) port (int): Port used to connect to etcd. srv_domain (str): Domain to search the SRV record for cluster autodiscovery. version_prefix (str): Url or version prefix in etcd url (default=/v2). read_timeout (int): max seconds to wait for a read. allow_redirect (bool): allow the client to connect to other nodes. protocol (str): Protocol used to connect to etcd. cert (mixed): If a string, the whole ssl client certificate; if a tuple, the cert and key file names. ca_cert (str): The ca certificate. If pressent it will enable validation. username (str): username for etcd authentication. password (str): password for etcd authentication. allow_reconnect (bool): allow the client to reconnect to another etcd server in the cluster in the case the default one does not respond. use_proxies (bool): we are using a list of proxies to which we connect, and don't want to connect to the original etcd cluster. expected_cluster_id (str): If a string, recorded as the expected UUID of the cluster (rather than learning it from the first request), reads will raise EtcdClusterIdChanged if they receive a response with a different cluster ID. per_host_pool_size (int): specifies maximum number of connections to pool by host. By default this will use up to 10 connections. lock_prefix (str): Set the key prefix at etcd when client to lock object. By default this will be use /_locks. Nz-Could not discover the etcd hosts from %s: %scSsd|||fS)Nz %s://%s:%d)protocolhostportrr?/home/ubuntu/.local/lib/python3.10/site-packages/etcd/client.pyuriszClient.__init__..uriz0List of hosts incompatible with allow_reconnect.zFA list of hosts to connect to was given, but reconnection not allowed?csg|] }jg|RqSr _protocol).0connselfr"rr! sz#Client.__init__..rmaxsizetimeout cert_filekey_fileca_certs cert_reqszHUsername provided without password, both are required for authenticationzHPassword provided without username, both are required for authentication num_poolsrzNew etcd client created for %sz Machines cache initialised to %sr)# _discover Exception_logerrorr$ isinstancetuple_machines_cache _base_urietcd EtcdExceptionpopexpected_cluster_idversion_prefix _read_timeout_allow_redirect _use_proxies_allow_reconnect _lock_prefixssl CERT_REQUIREDusernamepasswordwarningurllib3 PoolManagerrdebugbase_urilistsetmachinesremove_version_cluster_version)r(rr srv_domainr> read_timeoutallow_redirectrcertca_certrFrGallow_reconnect use_proxiesr=per_host_pool_size lock_prefixekwrr'r!__init__3sxF            zClient.__init__c CsLt|jj|j|jd||j|jdj d}|d|_ |d|_ dS)zF Sets the version information provided by the server. z/versionheadersr+redirectutf-8 etcdserver etcdclusterN) jsonloadsrrequest_MGETr9 _get_headersrTrUdatadecoderQrR)r( version_inforrr!_set_version_infos  zClient._set_version_infocCsdd|}tj|d}g}|D]}||jjdd|jfqt d|t |s.t dt |S)Nz _etcd._tcp.{}SRVT)omit_final_dotzFound %sz0The SRV record is present but no host were found) formatdnsresolverqueryappendtargetto_textr r4rKlen ValueErrorr7)r(domainsrv_nameanswershostsanswerrrr!r2s  zClient._discovercCs4|jdurz|jWdStyYdSwdS)zClean up open connectionsN)rclearReferenceErrorr(rrr!__del__s  zClient.__del__cC|jS)z*URI used by the client to connect to etcd.)r9rrrr!rLzClient.base_uricCst|jjddS)zNode to connect etcd.:r)rr9netlocsplitrrrr!rsz Client.hostcCstt|jjddS)zPort to connect etcd.rr-)intrr9rrrrrr!r sz Client.portcCr)zProtocol used to connect etcd.r#rrrr!rrzClient.protocolcCr)zMax seconds to wait for a read.)r?rrrr!rT rzClient.read_timeoutcCr)z+Allow the client to connect to other nodes.)r@rrrr!rUrzClient.allow_redirectcCr)z6Get the key prefix at etcd when client to lock object.)rCrrrr!r[rzClient.lock_prefixc Csz2|j|jd}|jj|j|||j|jd}dd||j d dD}t d||WStttjfyk}z)t d|j|j||jrb|jd |_t d |j|jWYd }~Std d }~ww) z Members of the cluster. Returns: list. str with all the nodes in the cluster. >>> print client.machines ['http://127.0.0.1:4001', 'http://127.0.0.1:4002'] z /machinesr_cSsg|]}|qSr)strip)r%noderrr!r)/sz#Client.machines..rb,zRetrieved list of machines: %sz,Failed to get list of machines from %s%s: %rrzRetrying on %sNzVCould not get the list of servers, maybe you provided the wrong host(s) to connect to?)r9r>rrgrhrirTrU_handle_server_responserjrkrr4rKrrsocketr5r8r<inforOr:r;)r(r"responserOr\rrr!rOs2    zClient.machinescCsbi|_z%||jd|jjd}t|}|dD] }||j|d<q|jWSt d)z A more structured view of peers in the cluster. Note that while we have an internal DS called _members, accessing the public property will call etcd. z/membersrbmembersidz@Could not get the members list, maybe the cluster has gone away?) _members api_executer>rhrjrkrerfr:r;)r(rjresmemberrrr!rDs    zClient.membersc Cs^zt||jd|jjd}|j|ddWSty.}zt d|d}~ww)z Returns: dict. the leader of the cluster. >>> print client.leader {"id":"ce2a822cea30bfca","name":"default","peerURLs":["http://localhost:2380","http://localhost:7001"],"clientURLs":["http://127.0.0.1:4001"]} z /stats/selfrb leaderInfoleaderzCannot get leader data: %sN) rerfrr>rhrjrkrr3r:r;)r(rr\rrr!rWs  z Client.leadercCs|S)zJ Returns: dict. the stats of the local server _statsrrrr!statsisz Client.statscC |dS)zD Returns: dict. the stats of the leader rrrrrr! leader_statsq zClient.leader_statscCr)zE Returns: dict. the stats of the kv store storerrrrr! store_statsyrzClient.store_statsr(c CsJ||jd||jjd}zt|WSttfy$t dw)z. Internal method to access the stats endpointsz/stats/rbz&Cannot parse json data in the response) rr>rhrjrkrerf TypeErrorrxr:r;)r(whatrjrrr!rs  z Client._statscC|js||jS)z" Version of etcd. )rQrmrrrr!versionszClient.versioncCr)z. Version of the etcd cluster. )rRrmrrrr!cluster_versionszClient.cluster_versioncCs |jdS)z$ REST key endpoint. z/keys)r>rrrr! key_endpoints zClient.key_endpointcCs(z||WdStjyYdSw)zm Check if a key is available in the cluster. >>> print 'key' in client True TF)getr:EtcdKeyNotFoundr(keyrrr! __contains__s  zClient.__contains__cCs|ds d|}|S)N/z/{}) startswithrprrrr! _sanitize_keys  zClient._sanitize_keyc Kstd|||||||}i}|dur||d<|dur!||d<|r.|r*tdd|d<|D]\}} ||jvrNt| tkrJ| rEdpFd||<q2| ||<q2|rT|j pV|j } d |vrb|d |} n|j |} |j | | |d } | | S) a9 Writes the value for a key, possibly doing atomic Compare-and-Swap Args: key (str): Key. value (object): value to set ttl (int): Time in seconds of expiration (optional). dir (bool): Set to true if we are writing a directory; default is false. append (bool): If true, it will post to append the new value to the dir, creating a sequential key. Defaults to false. Other parameters modifying the write method are accepted: prevValue (str): compare key to this value, and swap only if corresponding (optional). prevIndex (int): modify key only if actual modifiedIndex matches the provided one (optional). prevExist (bool): If false, only create key; if true, only update key. refresh (bool): since 2.3.0, If true, only update the ttl, prev key must existed(prevExist=True). Returns: client.EtcdResult >>> print client.write('/key', 'newValue', ttl=60, prevExist=False).value 'newValue' z,Writing %s to key %s ttl=%s dir=%s append=%sNvaluettlz&Cannot create a directory with a valuetruedirfalse _endpointparams)r4rKrr:r;items_comparison_conditionstypebool_MPOST_MPUTrr_result_from_response) r(rrrrrtkwdargsrkvmethodpathrrrr!writes6!      z Client.writecKs"d|d<|jd|d|dd|S)a[ (Since 2.3.0) Refresh the ttl of a key without notifying watchers. Keys in etcd can be refreshed without notifying watchers, this can be achieved by setting the refresh to true when updating a TTL You cannot update the value of a key when refreshing it @see: https://github.com/coreos/etcd/blob/release-2.3/Documentation/api.md#refreshing-key-ttl Args: key (str): Key. ttl (int): Time in seconds of expiration (optional). Other parameters modifying the write method are accepted as `EtcdClient.write`. TrN)rrrrrr)r(rrrrrr!rszClient.refreshcCsJtd|j|j|j|jdd}|js|j|d<|j|j|jfi|S)a Updates the value for a key atomically. Typical usage would be: c = etcd.Client() o = c.read("/somekey") o.value += 1 c.update(o) Args: obj (etcd.EtcdResult): The object that needs updating. zUpdating %s to %s.T)rrrr)r4rKrrrr modifiedIndexr)r(objrrrr!update s  z Client.updatecKstd||||}i}|D] \}}||jvr2t|tkr*|r%dp&d||<q|dur2|||<q|dd}|j|j ||j ||d}| |S)a Returns the value of the key 'key'. Args: key (str): Key. Recognized kwd args recursive (bool): If you should fetch recursively a dir wait (bool): If we should wait and return next time the key is changed waitIndex (int): The index to fetch results from. sorted (bool): Sort the output keys (alphanumerically) timeout (int): max seconds to wait for a read. Returns: client.EtcdResult (or an array of client.EtcdResult if a subtree is queried) Raises: KeyValue: If the key doesn't exists. urllib3.exceptions.TimeoutError: If timeout is reached. >>> print client.get('/key').value 'value' z$Issuing read for key %s with args %srrNr+rr+) r4rKrr _read_optionsrrrrrrhr)r(rrrrrr+rrrr!read&s      z Client.readcKstd||||||}i}|dur|rdpd|d<|dur(|r$dp%d|d<|jD] }||vr7||||<q+td||j|j||j|d}||S) a Removed a key from etcd. Args: key (str): Key. recursive (bool): if we want to recursively delete a directory, set it to true dir (bool): if we want to delete a directory, set it to true prevValue (str): compare key to this value, and swap only if corresponding (optional). prevIndex (int): modify key only if actual modifiedIndex matches the provided one (optional). Returns: client.EtcdResult Raises: KeyValue: If the key doesn't exists. >>> print client.delete('/key').key '/key' z-Deleting %s recursive=%s dir=%s extra args=%sNrrrrzCalculated params = %sr)r4rKr_del_conditionsrr_MDELETEr)r(rrrrkwdsrrrrr!deleteXs$     z Client.deletecKs|jd|||d|jS)a Remove specified key from etcd and return the corresponding value. Args: key (str): Key. recursive (bool): if we want to recursively delete a directory, set it to true dir (bool): if we want to delete a directory, set it to true prevValue (str): compare key to this value, and swap only if corresponding (optional). prevIndex (int): modify key only if actual modifiedIndex matches the provided one (optional). Returns: client.EtcdResult Raises: KeyValue: If the key doesn't exists. >>> print client.pop('/key').value 'value' )rrrNr)r _prev_node)r(rrrrrrr!r<sz Client.popcCs|j||||dS)a Atomic test & set operation. It will check if the value of 'key' is 'prev_value', if the the check is correct will change the value for 'key' to 'value' if the the check is false an exception will be raised. Args: key (str): Key. value (object): value to set prev_value (object): previous value. ttl (int): Time in seconds of expiration (optional). Returns: client.EtcdResult Raises: ValueError: When the 'prev_value' is not the current value. >>> print client.test_and_set('/key', 'new', 'old', ttl=60).value 'new' )r rr)r(rr prev_valuerrrr! test_and_setszClient.test_and_setcCs|j|||dS)ao Compatibility: sets the value of the key 'key' to the value 'value' Args: key (str): Key. value (object): value to set ttl (int): Time in seconds of expiration (optional). Returns: client.EtcdResult Raises: etcd.EtcdException: when something weird goes wrong. )rr)r(rrrrrr!rNsz Client.setcCs ||S)a Returns the value of the key 'key'. Args: key (str): Key. Returns: client.EtcdResult Raises: KeyError: If the key doesn't exists. >>> print client.get('/key').value 'value' )rrrrr!rs z Client.getcCs8td|||r|j|d|||dS|j|d||dS)a Blocks until a new event has been received, starting at index 'index' Args: key (str): Key. index (int): Index to start from. timeout (int): max seconds to wait for a read. Returns: client.EtcdResult Raises: KeyValue: If the key doesn't exist. etcd.EtcdWatchTimedOut: If timeout is reached. >>> print client.watch('/key').value 'value' z!About to wait on key %s, index %sT)rrr+r)rr+r)r4rKr)r(rindexr+rrrr!watchs  z Client.watchccs,|} |j||d|d}|jd}|Vq)a Generator that will yield changes from a key. Note that this method will block forever until an event is generated. Args: key (str): Key to subcribe to. index (int): Index from where the changes will be received. Yields: client.EtcdResult >>> for event in client.eternal_watch('/subcription_key'): ... print event.value ... value1 value2 Tr)rr+rr-)rr)r(rrr local_indexrrrr! eternal_watchs zClient.eternal_watchcOtd)Nz*Lock primitives were removed from etcd 2.0NotImplementedError)r(argskwargsrrr!get_lockszClient.get_lockcCr)Nz.Election primitives were removed from etcd 2.0rrrrr!election!szClient.electionc Cs|j}z t|d}Wntttfy#}ztd|d}~wwztj di|}|j dkr5d|_ | ||WSt yO}ztd|d}~ww)z, Creates an EtcdResult from json dictionary rbz&Server response was not valid JSON: %rNTz$Unable to decode server response: %rr)rjrerfrkrrx UnicodeErrorr:r; EtcdResultstatusnewKey parse_headersr3)r(r raw_responserr\rrrr!r%s*  zClient._result_from_responsecCsTtd|jz|j}Wnty!tdtjd|dwtd||S)zA Selects the next server in the list, refreshes the server list. z7Selection next machine in cache. Available machines: %sz,Machines cache is empty, no machines to try.zNo more machines in the clustercausezSelected new etcd server %s) r4rKr8r< IndexErrorr5r:EtcdConnectionFailedr)r(rmachrrr! _next_server7s   zClient._next_servercstdfdd }|S)Nc sd}|dur |j}|dkrd}|dstd|sd}z|||||d}|||j}Wnytttjfy}zOt |t rZ| ddkrZt |t rZt dtjd ||d t d |j||jrvt d |j|d |_d }d}nt dtjd||d WYd}~nd}~wtjy}zt |d}~wt d|r|js|j|_|j|j|r||S)NFrrzPath does not start with /rrrzWatch timed out.zWatch timed out: %rrzRequest to server %s failed: %rz1Reconnection allowed, looking for another server.Tz!Reconnection disabled, giving up.z#Connection to etcd failed due to %rz'Unexpected request failure, re-raising.)rTrrx_check_cluster_idrjrrrr5r6dictrrr4rKr:EtcdWatchTimedOutr9rBrrrEtcdClusterIdChangedrH exceptionrArOr8rPr) r(rrrr+rsome_request_failed_r\payloadrr!wrapperFsj           8z%Client._wrap_request..wrapperNNr)rrrrr! _wrap_requestEsEzClient._wrap_requestc Cs|j|}||jks||jkr|jj|||||j|ddS||jks)||jkr:|jj ||||d|j|ddSt d |)z Executes the query. F)r+fieldsrar`preload_content)rr+encode_multipartrar`rzHTTP method {} not supported) r9rhrrrgrUrirrrequest_encode_bodyr:r;rp)r(rrrr+urlrrr!rs2   zClient.api_executec Cs@|j|}t|}|}d|d<|jj|||||j|ddS)Nzapplication/jsonz Content-TypeF)bodyr+rar`r)r9redumpsrirurlopenrU)r(rrrr+r json_payloadr`rrr!api_execute_jsons  zClient.api_execute_jsoncCs\|d}|stddS|jo||jk}|j}||_|r,|jtd||dS)Nzx-etcd-cluster-idz*etcd response did not contain a cluster IDz.The UUID of the cluster changed from {} to {}.) getheaderr4rHr=rr~r:rrp)r(r cluster_id id_changedold_expected_cluster_idrrr!rs   zClient._check_cluster_idc Csh|jdvr|S|jd}z t|}|j|d<Wnttfy+dt|d}Ynwtj |dS)z Handles the server response )rrbrz Bad response)messagerN) rrjrkrerfrrxstrr: EtcdErrorhandle)r(rresprrrr!rs    zClient._handle_server_responsecCs.|jr|jrd|j|jf}tj|dSiS)Nr) basic_auth)rFrGjoinrI make_headers)r( credentialsrrr!ris  zClient._get_headers)rrNrrTrNNNNFFNrrr)NFFr)N)NNN)9__name__ __module__ __qualname____doc__rhrrrrNrrrrr^rmr2rpropertyrLrr rrTrUr[rOrrrrrrrrrrrrrrrrr<rrrrrrrrrrrrrrirrrr!r#s "         )          A 2 0     I    r)rlogging http.clientr ImportErrorhttplibrrIurllib3.exceptionsrrrerD dns.resolverrq functoolsrr:r urllib.parse getLoggerr r4objectrrrrr!s.