"""Common functionality for implementing backends."""
# Some utilities for handling tagging paths.
#
# The idea here is that etcd3 only supports straight-up prefix
# searches, but we do not want to get "/a/b/c" when listing "/a/"
# (just "/a/b"). Therefore we prepend all paths with the number of
# slashes they contain, making standard prefix search non-recursive as
# suggested by etcd's documentation. The recursive behaviour can
# always be restored by doing separate searches per recursion level.
[docs]
def depth_of_path(path: str) -> int:
"""
Get the depth of a path, this is the number of "/" in it.
:return: the depth
"""
return path.count("/")
def _tag_depth(path: str, depth=None) -> bytes:
"""
Add depth tag to path.
:param path: starting with /
:param depth: to add, number of / by default
:return: tag
"""
if not isinstance(path, str):
raise TypeError("Path must be a string!")
if not path or path[0] != "/":
raise ValueError("Path must start with /!")
if depth is None:
depth = depth_of_path(path)
# This returns a bytes object, because that's what this is natively in
# etcd3 (any byte sequence can be a key). This way tagged and untagged
# paths also have different types, and it's harder to mix them up.
return f"{depth}{path}".encode("utf-8")
def _untag_depth(tagged_path: bytes) -> str:
"""
Remove depth from tag.
:param tagged_path: tagged path
:return: path
"""
if not isinstance(tagged_path, bytes):
raise ValueError("Tagged path must be bytes!")
# Cut from first '/'
slash_ix = tagged_path.index(b"/")
if slash_ix is None:
return tagged_path.decode("utf-8")
return tagged_path[slash_ix:].decode("utf-8")
def _check_path(path: str) -> None:
if path and path[-1] == "/":
raise ValueError("Path should not have a trailing '/'!")
[docs]
class ConfigCollision(RuntimeError):
"""Exception generated if key to create already exists."""
def __init__(self, path: str, message: str):
"""Instantiate the exception."""
self.path = path
super().__init__(message)
[docs]
class ConfigVanished(RuntimeError):
"""Exception generated if key to update that does not exist."""
def __init__(self, path: str, message: str):
"""Instantiate the exception."""
self.path = path
super().__init__(message)