# etcd3gw.client

# Source code for etcd3gw.client

    #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    #    not use this file except in compliance with the License. You may obtain
    #    a copy of the License at
    #
    #         http://www.apache.org/licenses/LICENSE-2.0
    #
    #    Unless required by applicable law or agreed to in writing, software
    #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    #    License for the specific language governing permissions and limitations
    #    under the License.

    import json
    import os
    import queue
    import threading
    import uuid

    import requests

    from etcd3gw import exceptions
    from etcd3gw.lease import Lease
    from etcd3gw.lock import Lock
    from etcd3gw.utils import _decode
    from etcd3gw.utils import _encode
    from etcd3gw.utils import _increment_last_byte
    from etcd3gw.utils import DEFAULT_TIMEOUT
    from etcd3gw import watch

    _SORT_ORDER = ['none', 'ascend', 'descend']
    _SORT_TARGET = ['key', 'version', 'create', 'mod', 'value']

    _EXCEPTIONS_BY_CODE = {
        requests.codes['internal_server_error']: exceptions.InternalServerError,
        requests.codes['service_unavailable']: exceptions.ConnectionFailedError,
        requests.codes['request_timeout']: exceptions.ConnectionTimeoutError,
        requests.codes['gateway_timeout']: exceptions.ConnectionTimeoutError,
        requests.codes['precondition_failed']: exceptions.PreconditionFailedError,
    }

    DEFAULT_API_PATH = os.getenv('ETCD3GW_API_PATH')

    class Etcd3Client(object):
        def __init__(self, host='localhost', port=2379, protocol="http",
                     ca_cert=None, cert_key=None, cert_cert=None, timeout=None,
                     api_path=DEFAULT_API_PATH):
            """Construct an client to talk to etcd3's grpc-gateway's /v3 HTTP API

            :param host:
            :param port:
            :param protocol:
            """
            self.host = host
            self.port = port
            self.protocol = protocol

            self.session = requests.Session()
            self.timeout = timeout
            if ca_cert is not None:
                self.session.verify = ca_cert
            if cert_cert is not None and cert_key is not None:
                self.session.cert = (cert_cert, cert_key)
            self._api_path = api_path

        @property
        def api_path(self):
            if self._api_path is not None:
                return self._api_path
            self._discover_api_path()
            return self._api_path

        @property
        def base_url(self):
            host = ('[' + self.host + ']' if (self.host.find(':') != -1)
                    else self.host)
            return self.protocol + '://' + host + ':' + str(self.port)

        def _discover_api_path(self):
            """Discover api version and set api_path

            """
            resp = self._request('get', self.base_url + '/version')
            try:
                version_str = resp['etcdserver']
            except KeyError:
                raise exceptions.ApiVersionDiscoveryFailedError(
                    'Malformed response from version API')

            try:
                version = tuple(int(part) for part in version_str.split('.', 2))
            except ValueError:
                raise exceptions.ApiVersionDiscoveryFailedError(
                    'Failed to parse etcd cluster version: %s' % version_str)

            # NOTE(tkajinam): https://etcd.io/docs/v3.5/dev-guide/api_grpc_gateway/
            #                 explains mapping between etcd version and available
            #                 api versions
            if version >= (3, 4):
                self._api_path = '/v3/'
            elif version >= (3, 3):
                self._api_path = '/v3beta/'
            else:
                self._api_path = '/v3alpha/'

        def get_url(self, path):
            """Construct a full url to the v3 API given a specific path

            :param path:
            :return: url
            """

            return self.base_url + self.api_path + path.lstrip("/")

        def _request(self, method, *args, **kwargs):
            """helper method for HTTP requests

            :param args:
            :param kwargs:
            :return: json response
            """
            try:
                resp = getattr(self.session, method)(*args, timeout=self.timeout,
                                                     **kwargs)
                if resp.status_code in _EXCEPTIONS_BY_CODE:
                    raise _EXCEPTIONS_BY_CODE[resp.status_code](
                        resp.text,
                        resp.reason
                    )
                if resp.status_code != requests.codes['ok']:
                    raise exceptions.Etcd3Exception(resp.text, resp.reason)
            except requests.exceptions.Timeout as ex:
                raise exceptions.ConnectionTimeoutError(str(ex))
            except requests.exceptions.ConnectionError as ex:
                raise exceptions.ConnectionFailedError(str(ex))
            return resp.json()

        def post(self, *args, **kwargs):
            """helper method for HTTP POST

            :param args:
            :param kwargs:
            :return: json response
            """
            return self._request('post', *args, **kwargs)

        def status(self):
            """Status gets the status of the etcd cluster member.

            :return: json response
            """
            return self.post(self.get_url("/maintenance/status"),
                             json={})

        def members(self):
            """Lists all the members in the cluster.

            :return: json response
            """
            result = self.post(self.get_url("/cluster/member/list"),
                               json={})
            return result['members']

        def lease(self, ttl=DEFAULT_TIMEOUT):
            """Create a Lease object given a timeout

            :param ttl: timeout
            :return: Lease object
            """
            result = self.post(self.get_url("/lease/grant"),
                               json={"TTL": ttl, "ID": 0})
            return Lease(int(result['ID']), client=self)

        def lock(self, id=None, ttl=DEFAULT_TIMEOUT):
            """Create a Lock object given an ID and timeout

            :param id: ID for the lock, creates a new uuid if not provided
            :param ttl: timeout
            :return: Lock object
            """
            if id is None:
                id = str(uuid.uuid4())
            return Lock(id, ttl=ttl, client=self)

        def create(self, key, value, lease=None):
            """Atomically create the given key only if the key doesn't exist.

            This verifies that the create_revision of a key equales to 0, then
            creates the key with the value.
            This operation takes place in a transaction.

            :param key: key in etcd to create
            :param value: value of the key
            :type value: bytes or string
            :param lease: lease to connect with, optional
            :returns: status of transaction, ``True`` if the create was
                      successful, ``False`` otherwise
            :rtype: bool
            """
            base64_key = _encode(key)
            base64_value = _encode(value)
            txn = {
                'compare': [{
                    'key': base64_key,
                    'result': 'EQUAL',
                    'target': 'CREATE',
                    'create_revision': 0
                }],
                'success': [{
                    'request_put': {
                        'key': base64_key,
                        'value': base64_value,
                    }
                }],
                'failure': []
            }
            if lease:
                txn['success'][0]['request_put']['lease'] = lease.id
            result = self.transaction(txn)
            if 'succeeded' in result:
                return result['succeeded']
            return False

        def put(self, key, value, lease=None):
            """Put puts the given key into the key-value store.

            A put request increments the revision of the key-value store
            and generates one event in the event history.

            :param key:
            :param value:
            :param lease:
            :return: boolean
            """
            payload = {
                "key": _encode(key),
                "value": _encode(value)
            }
            if lease:
                payload['lease'] = lease.id
            self.post(self.get_url("/kv/put"), json=payload)
            return True

        def get(self, key, metadata=False, sort_order=None,
                sort_target=None, **kwargs):
            """Range gets the keys in the range from the key-value store.

            :param key:
            :param metadata:
            :param sort_order: 'ascend' or 'descend' or None
            :param sort_target: 'key' or 'version' or 'create' or 'mod' or 'value'
            :param kwargs:
            :return:
            """
            try:
                order = 0
                if sort_order:
                    order = _SORT_ORDER.index(sort_order)
            except ValueError:
                raise ValueError('sort_order must be one of "ascend" or "descend"')

            try:
                target = 0
                if sort_target:
                    target = _SORT_TARGET.index(sort_target)
            except ValueError:
                raise ValueError('sort_target must be one of "key", '
                                 '"version", "create", "mod" or "value"')

            payload = {
                "key": _encode(key),
                "sort_order": order,
                "sort_target": target,
            }
            payload.update(kwargs)
            result = self.post(self.get_url("/kv/range"),
                               json=payload)
            if 'kvs' not in result:
                return []

            if metadata:
                def value_with_metadata(item):
                    item['key'] = _decode(item['key'])
                    value = _decode(item.pop('value', ''))
                    return value, item
                return [value_with_metadata(item) for item in result['kvs']]

            return [_decode(item.get('value', '')) for item in result['kvs']]

        def get_all(self, sort_order=None, sort_target='key'):
            """Get all keys currently stored in etcd.

            :returns: sequence of (value, metadata) tuples
            """
            return self.get(
                key=_encode(b'\0'),
                metadata=True,
                sort_order=sort_order,
                sort_target=sort_target,
                range_end=_encode(b'\0'),
            )

        def get_prefix(self, key_prefix, sort_order=None, sort_target=None):
            """Get a range of keys with a prefix.

            :param sort_order: 'ascend' or 'descend' or None
            :param key_prefix: first key in range

            :returns: sequence of (value, metadata) tuples
            """
            return self.get(key_prefix,
                            metadata=True,
                            range_end=_encode(_increment_last_byte(key_prefix)),
                            sort_order=sort_order,
                            sort_target=sort_target)

        def replace(self, key, initial_value, new_value):
            """Atomically replace the value of a key with a new value.

            This compares the current value of a key, then replaces it with a new
            value if it is equal to a specified value. This operation takes place
            in a transaction.

            :param key: key in etcd to replace
            :param initial_value: old value to replace
            :type initial_value: bytes or string
            :param new_value: new value of the key
            :type new_value: bytes or string
            :returns: status of transaction, ``True`` if the replace was
                      successful, ``False`` otherwise
            :rtype: bool
            """
            base64_key = _encode(key)
            base64_initial_value = _encode(initial_value)
            base64_new_value = _encode(new_value)
            txn = {
                'compare': [{
                    'key': base64_key,
                    'result': 'EQUAL',
                    'target': 'VALUE',
                    'value': base64_initial_value
                }],
                'success': [{
                    'request_put': {
                        'key': base64_key,
                        'value': base64_new_value,
                    }
                }],
                'failure': []
            }
            result = self.transaction(txn)
            if 'succeeded' in result:
                return result['succeeded']
            return False

        def delete(self, key, **kwargs):
            """DeleteRange deletes the given range from the key-value store.

            A delete request increments the revision of the key-value store and
            generates a delete event in the event history for every deleted key.

            :param key:
            :param kwargs:
            :return:
            """
            payload = {
                "key": _encode(key),
            }
            payload.update(kwargs)

            result = self.post(self.get_url("/kv/deleterange"),
                               json=payload)
            if 'deleted' in result:
                return True
            return False

        def delete_prefix(self, key_prefix):
            """Delete a range of keys with a prefix in etcd."""
            return self.delete(
                key_prefix, range_end=_encode(_increment_last_byte(key_prefix)))

        def transaction(self, txn):
            """Txn processes multiple requests in a single transaction.

            A txn request increments the revision of the key-value store and
            generates events with the same revision for every completed request.
            It is not allowed to modify the same key several times within one txn.

            :param txn:
            :return:
            """
            return self.post(self.get_url("/kv/txn"),
                             data=json.dumps(txn))

        def watch(self, key, **kwargs):
            """Watch a key.

            :param key: key to watch

            :returns: tuple of ``events_iterator`` and ``cancel``.
                      Use ``events_iterator`` to get the events of key changes
                      and ``cancel`` to cancel the watch request
            """
            event_queue = queue.Queue()

            def callback(event):
                event_queue.put(event)

            w = watch.Watcher(self, key, callback, **kwargs)
            canceled = threading.Event()

            def cancel():
                canceled.set()
                event_queue.put(None)
                w.stop()

            def iterator():
                while not canceled.is_set():
                    event = event_queue.get()
                    if event is None:
                        canceled.set()
                    if not canceled.is_set():
                        yield event

            return iterator(), cancel

        def watch_prefix(self, key_prefix, **kwargs):
            """The same as ``watch``, but watches a range of keys with a prefix."""
            kwargs['range_end'] = \
                _increment_last_byte(key_prefix)
            return self.watch(key_prefix, **kwargs)

        def watch_once(self, key, timeout=None, **kwargs):
            """Watch a key and stops after the first event.

            :param key: key to watch
            :param timeout: (optional) timeout in seconds.
            :returns: event
            """
            event_queue = queue.Queue()

            def callback(event):
                event_queue.put(event)

            w = watch.Watcher(self, key, callback, **kwargs)
            try:
                return event_queue.get(timeout=timeout)
            except queue.Empty:
                raise exceptions.WatchTimedOut()
            finally:
                w.stop()

        def watch_prefix_once(self, key_prefix, timeout=None, **kwargs):
            """Watches a range of keys with a prefix, similar to watch_once"""
            kwargs['range_end'] = \
                _increment_last_byte(key_prefix)
            return self.watch_once(key_prefix, timeout=timeout, **kwargs)

    def client(host='localhost', port=2379,
               ca_cert=None, cert_key=None, cert_cert=None,
               timeout=None, protocol="http", api_path=DEFAULT_API_PATH):
        """Return an instance of an Etcd3Client."""
        return Etcd3Client(host=host,
                           port=port,
                           ca_cert=ca_cert,
                           cert_key=cert_key,
                           cert_cert=cert_cert,
                           timeout=timeout,
                           api_path=api_path,
                           protocol=protocol)

this page last updated: 2024-12-19 09:58:08

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.