from __future__ import annotations
from consul.callback import CB
[docs]
class KV:
"""
The KV endpoint is used to expose a simple key/value store. This can be
used to store service configurations or other meta data in a simple
way.
"""
def __init__(self, agent) -> None:
self.agent = agent
[docs]
def get(
self,
key,
index=None,
recurse: bool = False,
wait=None,
token: str | None = None,
consistency=None,
keys: bool = False,
separator=None,
dc=None,
connections_timeout=None,
):
"""
Returns a tuple of (*index*, *value[s]*)
*index* is the current Consul index, suitable for making subsequent
calls to wait for changes since this query was last run.
*wait* the maximum duration to wait (e.g. '10s') to retrieve
a given index. this parameter is only applied if *index* is also
specified. the wait time by default is 5 minutes.
*token* is an optional `ACL token`_ to apply to this request.
*keys* is a boolean which, if True, says to return a flat list of
keys without values or other metadata. *separator* can be used
with *keys* to list keys only up to a given separator character.
*dc* is the optional datacenter that you wish to communicate with.
If None is provided, defaults to the agent's datacenter.
The *value* returned is for the specified key, or if *recurse* is
True a list of *values* for all keys with the given prefix is
returned.
Each *value* looks like this::
{
"CreateIndex": 100,
"ModifyIndex": 200,
"LockIndex": 200,
"Key": "foo",
"Flags": 0,
"Value": "bar",
"Session": "adf4238a-882b-9ddc-4a9d-5b6758e4159e"
}
Note, if the requested key does not exists *(index, None)* is
returned. It's then possible to long poll on the index for when the
key is created.
"""
assert not key.startswith("/"), "keys should not start with a forward slash"
params = []
if index:
params.append(("index", index))
if wait:
params.append(("wait", wait))
if recurse:
params.append(("recurse", "1"))
dc = dc or self.agent.dc
if dc:
params.append(("dc", dc))
if keys:
params.append(("keys", True))
if separator:
params.append(("separator", separator))
consistency = consistency or self.agent.consistency
if consistency in ("consistent", "stale"):
params.append((consistency, "1"))
one = False
decode: bool | str = False
if not keys:
decode = "Value"
if not recurse and not keys:
one = True
http_kwargs = {}
if connections_timeout:
http_kwargs["connections_timeout"] = connections_timeout
headers = self.agent.prepare_headers(token)
return self.agent.http.get(
CB.json(index=True, decode=decode, one=one), f"/v1/kv/{key}", params=params, headers=headers, **http_kwargs
)
[docs]
def put(
self,
key,
value,
cas=None,
flags=None,
acquire=None,
release=None,
token: str | None = None,
dc=None,
connections_timeout=None,
):
"""
Sets *key* to the given *value*.
*value* can either be None (useful for marking a key as a
directory) or any string type, including binary data (e.g. a
msgpack'd data structure)
The optional *cas* parameter is used to turn the PUT into a
Check-And-Set operation. This is very useful as it allows clients
to build more complex syncronization primitives on top. If the
index is 0, then Consul will only put the key if it does not
already exist. If the index is non-zero, then the key is only set
if the index matches the ModifyIndex of that key.
An optional *flags* can be set. This can be used to specify an
unsigned value between 0 and 2^64-1.
*acquire* is an optional session_id. if supplied a lock acquisition
will be attempted.
*release* is an optional session_id. if supplied a lock release
will be attempted.
*token* is an optional `ACL token`_ to apply to this request. If
the token's policy is not allowed to write to this key an
*ACLPermissionDenied* exception will be raised.
*dc* is the optional datacenter that you wish to communicate with.
If None is provided, defaults to the agent's datacenter.
The return value is simply either True or False. If False is
returned, then the update has not taken place.
"""
assert not key.startswith("/"), "keys should not start with a forward slash"
assert value is None or isinstance(value, (str, bytes)), "value should be None or a string / binary data"
params = []
if cas is not None:
params.append(("cas", cas))
if flags is not None:
params.append(("flags", flags))
if acquire:
params.append(("acquire", acquire))
if release:
params.append(("release", release))
dc = dc or self.agent.dc
if dc:
params.append(("dc", dc))
http_kwargs = {}
if connections_timeout:
http_kwargs["connections_timeout"] = connections_timeout
headers = self.agent.prepare_headers(token)
return self.agent.http.put(
CB.json(), f"/v1/kv/{key}", params=params, headers=headers, data=value, **http_kwargs
)
[docs]
def delete(self, key, recurse=None, cas=None, token: str | None = None, dc=None, connections_timeout=None):
"""
Deletes a single key or if *recurse* is True, all keys sharing a
prefix.
*cas* is an optional flag is used to turn the DELETE into a
Check-And-Set operation. This is very useful as a building block
for more complex synchronization primitives. Unlike PUT, the index
must be greater than 0 for Consul to take any action: a 0 index
will not delete the key. If the index is non-zero, the key is only
deleted if the index matches the ModifyIndex of that key.
*token* is an optional `ACL token`_ to apply to this request. If
the token's policy is not allowed to delete to this key an
*ACLPermissionDenied* exception will be raised.
*dc* is the optional datacenter that you wish to communicate with.
If None is provided, defaults to the agent's datacenter.
"""
assert not key.startswith("/"), "keys should not start with a forward slash"
params = []
if recurse:
params.append(("recurse", "1"))
if cas is not None:
params.append(("cas", cas))
dc = dc or self.agent.dc
if dc:
params.append(("dc", dc))
http_kwargs = {}
if connections_timeout:
http_kwargs["connections_timeout"] = connections_timeout
headers = self.agent.prepare_headers(token)
return self.agent.http.delete(CB.json(), f"/v1/kv/{key}", params=params, headers=headers, **http_kwargs)