uncloud-mravi/ucloud/common/counters.py

from .etcd_wrapper import Etcd3Wrapper


def increment_etcd_counter(etcd_client: Etcd3Wrapper, key):
    kv = etcd_client.get(key)

    if kv:
        counter = int(kv.value)
        counter = counter + 1
    else:
        counter = 1

    etcd_client.put(key, str(counter))
    return counter


def get_etcd_counter(etcd_client: Etcd3Wrapper, key):
    kv = etcd_client.get(key)
    if kv:
        return int(kv.value)
    return None