hyperion.ports.keyval

hyperion.ports.keyval

Port: key-value store abstraction.

Abstract :class:KeyValueStore base, the :data:CompressionType alias and the :func:is_valid_compression_type guard. If you need a store with a TTL, look at :class:hyperion.ports.cache.Cache instead. Concrete adapters (DynamoDBStore, InMemoryStore, FilesystemStore) live in hyperion.adapters.keyval.*.

KeyValueStore

KeyValueStore(prefix=None, compression=None)

Bases: ABC, Iterable[str]

A key-value store.

Initialize the store.

Parameters:

Name	Type	Description	Default
prefix	str | None	A prefix to add to all keys.	None
compression	CompressionType | None	The compression algorithm to use.	None

Source code in hyperion/ports/keyval.py

def __init__(self, prefix: str | None = None, compression: CompressionType | None = None) -> None:
    """Initialize the store.

    Args:
        prefix: A prefix to add to all keys.
        compression: The compression algorithm to use.
    """
    self.prefix = prefix
    self.compression = compression

get

get(key)

Gets a value from the store.

Source code in hyperion/ports/keyval.py

def get(self, key: str) -> str | None:
    """Gets a value from the store."""
    if value := self._get_raw(self._key(key)):
        return self._decompress(value)
    return None

set

set(key, value)

Sets a value in the store.

Source code in hyperion/ports/keyval.py

def set(self, key: str, value: str) -> None:
    """Sets a value in the store."""
    return self._set_raw(self._key(key), self._compress(value))

keys

keys(match='*')

Iterate all keys available in the store that match the given expression.

Nothing fancy is supported, UNIX-style fnmatch is performed.

Parameters:

Name	Type	Description	Default
match	str	The expression to match.	'*'

Returns:

Type	Description
Iterable[str]	Iterable[str]: The keys that match the expression

Source code in hyperion/ports/keyval.py

def keys(self, match: str = "*") -> Iterable[str]:
    """Iterate all keys available in the store that match the given expression.

    Nothing fancy is supported, UNIX-style fnmatch is performed.

    Args:
        match: The expression to match.

    Returns:
        Iterable[str]: The keys that match the expression
    """
    for key in self._iter_all_keys():
        if fnmatch(key, match):
            yield key

delete

delete(key)

Delete a value from the store.

Parameters:

Name	Type	Description	Default
key	str	The key to delete.	required

Source code in hyperion/ports/keyval.py

def delete(self, key: str) -> None:
    """Delete a value from the store.

    Args:
        key: The key to delete.
    """
    return self._delete_raw(self._key(key))

exists

exists(key)

Returns whether a key exists in the store.

Parameters:

Name	Type	Description	Default
key	str	The key to check.	required

Source code in hyperion/ports/keyval.py

def exists(self, key: str) -> bool:
    """Returns whether a key exists in the store.

    Args:
        key: The key to check.
    """
    return key in self.keys()