valkey_dict package

Submodules

valkey_dict.core module

Valkey Dict module.

class valkey_dict.core.ValkeyDict(namespace='main', expire=None, preserve_expiration=False, valkey=None, raise_key_error_delete=False, **valkey_kwargs)

Bases: object

Python dictionary with Valkey as backend.

With support for advanced features, such as custom data types, pipelining, and key expiration.

This class provides a dictionary-like interface that interacts with a Valkey database, allowing for efficient storage and retrieval of key-value pairs. It supports various data types, including strings, integers, floats, lists, dictionaries, tuples, sets, and user-defined types. The class leverages the power of Valkey pipelining to minimize network round-trip time, latency, and I/O load, thereby optimizing performance for batch operations. Additionally, it allows for the management of key expiration through the use of context managers.

The ValkeyDict class is designed to be analogous to a standard Python dictionary while providing enhanced functionality, such as support for a wider range of data types and efficient batch operations. It aims to offer a seamless and familiar interface for developers familiar with Python dictionaries, enabling a smooth transition to a Valkey-backed data store.

Extendable Types: You can extend ValkeyDict by adding or overriding encoding and decoding functions. This functionality enables various use cases, such as managing encrypted data in Valkey, To implement this, simply create and register your custom encoding and decoding functions. By delegating serialization to valkey-dict, reduce complexity and have simple code in the codebase.

__init__(namespace='main', expire=None, preserve_expiration=False, valkey=None, raise_key_error_delete=False, **valkey_kwargs)

Initialize a ValkeyDict instance.

Init the ValkeyDict instance.

Parameters:

• namespace (str) – A prefix for keys stored in Valkey.

• expire (Union[int, timedelta, None], optional) – Expiration time for keys.

• preserve_expiration (Optional[bool], optional) – Preserve expiration on key updates.

• valkey (Optional[StrictValkey[Any]], optional) – A Valkey connection instance.

• raise_key_error_delete (bool) – Enable strict Python dict behavior raise if key not found when deleting.

• **valkey_kwargs (Any) – Additional kwargs for Valkey connection if not provided.

chain_del(iterable)

Delete a value from the ValkeyDict using a chain of keys.

Parameters:

iterable (List[str]) – A list of keys representing the chain.

Return type:

None

chain_get(iterable)

Get a value from the ValkeyDict using a chain of keys.

Parameters:

iterable (List[str]) – A list of keys representing the chain.

Returns:

The value associated with the chain of keys.

Return type:

Any

chain_set(iterable, v)

Set a value in the ValkeyDict using a chain of keys.

Parameters:

• iterable (List[str]) – A list of keys representing the chain.

• v (Any) – The value to be set.

Return type:

None

clear()

Remove all key-value pairs from the ValkeyDict in one batch operation using pipelining.

This method mimics the behavior of the clear method from a standard Python dictionary. Valkey pipelining is employed to group multiple commands into a single request, minimizing network round-trip time, latency, and I/O load, thereby enhancing the overall performance.

Return type:

None

copy()

Create a shallow copy of the ValkeyDict and return it as a standard Python dictionary.

This method is analogous to the copy method of a standard Python dictionary

Returns:

A shallow copy of the ValkeyDict as a standard Python dictionary.

Return type:

dict

Note

does not create a new ValkeyDict instance.

decoding_registry: Dict[str, Callable[[str], Any]] = {'Decimal': <class 'decimal.Decimal'>, 'NoneType': <function <lambda>>, 'OrderedDict': <function <lambda>>, 'UUID': <class 'uuid.UUID'>, 'bool': <function <lambda>>, 'bytes': <function b64decode>, 'complex': <function <lambda>>, 'date': <built-in method fromisoformat of type object>, 'datetime': <built-in method fromisoformat of type object>, 'defaultdict': <function <lambda>>, 'dict': <function decode_json>, 'float': <class 'float'>, 'frozenset': <function <lambda>>, 'int': <class 'int'>, 'list': <function decode_json>, 'set': <function _decode_set>, 'str': <class 'str'>, 'time': <built-in method fromisoformat of type object>, 'timedelta': <function <lambda>>, 'tuple': <function _decode_tuple>}

encoding_registry: Dict[str, Callable[[Any], str]] = {'OrderedDict': <function <lambda>>, 'bytes': <function <lambda>>, 'complex': <function <lambda>>, 'date': <method 'isoformat' of 'datetime.date' objects>, 'datetime': <method 'isoformat' of 'datetime.datetime' objects>, 'defaultdict': <function <lambda>>, 'dict': <function encode_json>, 'frozenset': <function <lambda>>, 'list': <function encode_json>, 'set': <function _encode_set>, 'time': <method 'isoformat' of 'datetime.time' objects>, 'timedelta': <function <lambda>>, 'tuple': <function _encode_tuple>}

expire_at(sec_epoch)

Context manager to set the expiration time for keys in the ValkeyDict.

Parameters:

sec_epoch (int, timedelta) – The expiration duration is set using either an integer or a timedelta.

Yields:

ContextManager – A context manager during which the expiration time is the time set.

Return type:

Iterator[None]

extends_type(class_type, encode=None, decode=None, encoding_method_name=None, decoding_method_name=None)

Extend ValkeyDict to support a custom type in the encode/decode mapping.

This method enables serialization of instances based on their type, allowing for custom types, specialized storage formats, and more. There are three ways to add custom types: 1. Have a class with an encode instance method and a decode class method. 2. Have a class and pass encoding and decoding functions, where encode converts the class instance to a string, and decode takes the string and recreates the class instance. 3. Have a class that already has serialization methods, that satisfies the: EncodeFuncType = Callable[[Any], str] DecodeFuncType = Callable[[str], Any]

custom_encode_method custom_decode_method

If no encoding or decoding function is provided, default to use the encode and decode methods of the class.

The encode method should be an instance method that converts the object to a string. The decode method should be a class method that takes a string and returns an instance of the class.

The method names for encoding and decoding can be changed by modifying the - custom_encode_method - custom_decode_method attributes of the ValkeyDict instance

Example

>>> class Person:
...     def __init__(self, name, age):
...         self.name = name
...         self.age = age
...
...     def encode(self) -> str:
...         return json.dumps(self.__dict__)
...
...     @classmethod
...     def decode(cls, encoded_str: str) -> 'Person':
...         return cls(**json.loads(encoded_str))
...
>>> valkey_dict.extends_type(Person)

Parameters:

• class_type (type) – The class __name__ will become the key for the encoding and decoding functions.

• encode (Optional[EncodeFuncType]) – function that encodes an object into a storable string format.

• decode (Optional[DecodeFuncType]) – function that decodes a string back into an object of class_type.

• encoding_method_name (str, optional) – Name of encoding method of the class for valkey-dict custom types.

• decoding_method_name (str, optional) – Name of decoding method of the class for valkey-dict custom types.

Raises:

NotImplementedError –

Return type:

None

Note

You can check for compliance of a class separately using the new_type_compliance method:

This method raises a NotImplementedError if either encode or decode is None and the class does not implement the corresponding method.

fromkeys(iterable, value=None)

Create a new ValkeyDict from an iterable of key-value pairs.

Create a new ValkeyDict with keys from the provided iterable and values set to the given value. This method is analogous to the fromkeys method of a standard Python dictionary, populating the ValkeyDict with the keys from the iterable and setting their corresponding values to the specified value.

Parameters:

• iterable (List[str]) – An iterable containing the keys to be added to the ValkeyDict.

• value (Optional[Any], optional) – The value to be assigned to each key in the ValkeyDict. Defaults to None.

Returns:

The current ValkeyDict instance,populated with the keys from the iterable and their corresponding values.

Return type:

ValkeyDict

get(key, default=None)

Return the value for the given key if it exists, otherwise return the default value.

Analogous to a dictionary’s get method.

Parameters:

• key (str) – The key to retrieve the value.

• default (Optional[Any], optional) – The value to return if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

get_ttl(key)

Get the Time To Live from Valkey.

Get the Time To Live (TTL) in seconds for a given key. If the key does not exist or does not have an associated expire, return None.

Parameters:

key (str) – The key for which to get the TTL.

Returns:

The TTL in seconds if the key exists and has an expiry set; otherwise, None.

Return type:

Optional[int]

get_valkey_info()

Retrieve information and statistics about the Valkey server.

Returns:

The information and statistics from the Valkey server in a dictionary.

Return type:

dict

items()

Return a list of key-value pairs (tuples) in the ValkeyDict, analogous to a dictionary’s items method.

Yields:

Iterator[Tuple[str, Any]] – A list of key-value pairs in the ValkeyDict.

Return type:

Iterator[Tuple[str, Any]]

key(search_term='')

Return the first value for search_term if it exists, otherwise return None.

Parameters:

search_term (str) – A search term to filter keys. Defaults to ‘’.

Returns:

The first key associated with the given search term.

Return type:

str

keys()

Return an Iterator of keys in the ValkeyDict, analogous to a dictionary’s keys method.

Returns:

A list of keys in the ValkeyDict.

Return type:

Iterator[str]

multi_chain_get(keys)

Get multiple values from the ValkeyDict using a chain of keys.

Parameters:

keys (List[str]) – A list of keys representing the chain.

Returns:

A list of values associated with the chain of keys.

Return type:

List[Any]

multi_del(key)

Delete multiple values from the ValkeyDict using a shared key prefix.

Parameters:

key (str) – The shared key prefix.

Returns:

The number of keys deleted.

Return type:

int

multi_dict(key)

Get a dictionary of key-value pairs from the ValkeyDict using a shared key prefix.

Parameters:

key (str) – The shared key prefix.

Returns:

A dictionary of key-value pairs associated with the key prefix.

Return type:

Dict[str, Any]

multi_get(key)

Get multiple values from the ValkeyDict using a shared key prefix.

Parameters:

key (str) – The shared key prefix.

Returns:

A list of values associated with the key prefix.

Return type:

List[Any]

new_type_compliance(class_type, encode_method_name=None, decode_method_name=None)

Check if a class complies with the required encoding and decoding methods.

Parameters:

• class_type (type) – The class to check for compliance.

• encode_method_name (str, optional) – Name of encoding method of the class for valkey-dict custom types.

• decode_method_name (str, optional) – Name of decoding method of the class for valkey-dict custom types.

Raises:

NotImplementedError – If the class does not implement the required methods when the respective check is True.

Return type:

None

next()

Get the next item in the iterator (alias for __next__).

Returns:

The next item in the iterator.

Return type:

str

pipeline()

Context manager to create a Valkey pipeline for batch operations.

Yields:

ContextManager – A context manager to create a Valkey pipeline batching all operations within the context.

Return type:

Iterator[None]

pop(key, default=<object object>)

Analogous to a dictionary’s pop method.

Remove the value associated with the given key and return it, or return the default value if the key is not found.

Parameters:

• key (str) – The key to remove the value.

• default (Optional[Any], optional) – The value to return if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

Raises:

KeyError – If the key is not found and no default value is provided.

popitem()

Remove and return a random (key, value) pair from the ValkeyDict as a tuple.

This method is analogous to the popitem method of a standard Python dictionary.

if dict_compliant set true stays true to In Python 3.7+, removes the last inserted item (LIFO order)

Returns:

A tuple containing a randomly chosen (key, value) pair.

Return type:

tuple

Raises:

KeyError – If ValkeyDict is empty.

setdefault(key, default_value=None)

Get value under key, and if not present set default value.

Return the value associated with the given key if it exists, otherwise set the value to the default value and return it. Analogous to a dictionary’s setdefault method.

Parameters:

• key (str) – The key to retrieve the value.

• default_value (Optional[Any], optional) – The value to set if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

to_dict()

Convert the ValkeyDict to a Python dictionary.

Returns:

A dictionary with the same key-value pairs as the ValkeyDict.

Return type:

Dict[str, Any]

update(dic)

Update the ValkeyDict with key-value pairs from the given mapping, analogous to a dictionary’s update method.

Parameters:

dic (Mapping[str, Any]) – A mapping containing key-value pairs to update the ValkeyDict.

Return type:

None

values()

Analogous to a dictionary’s values method.

Return a list of values in the ValkeyDict,

Yields:

List[Any] – A list of values in the ValkeyDict.

Return type:

Iterator[Any]

valkey_dict.python_dict module

Python Valkey Dict module.

class valkey_dict.python_dict.PythonValkeyDict(namespace='main', expire=None, preserve_expiration=False, valkey=None, **valkey_kwargs)

Bases: ValkeyDict

Python dictionary with Valkey as backend.

With support for advanced features, such as custom data types, pipelining, and key expiration.

This class focuses on having one-to-on behavior of a dictionary while using Valkey as storage layer, allowing for efficient storage and retrieval of key-value pairs. It supports various data types, including strings, integers, floats, lists, dictionaries, tuples, sets, and user-defined types. The class leverages the power of Valkey pipelining to minimize network round-trip time, latency, and I/O load, thereby optimizing performance for batch operations. Additionally, it allows for the management of key expiration through the use of context managers.

The ValkeyDict class is designed to be analogous to a standard Python dictionary while providing enhanced functionality, such as support for a wider range of data types and efficient batch operations. It aims to offer a seamless and familiar interface for developers familiar with Python dictionaries, enabling a smooth transition to a Valkey-backed data store.

Extendable Types: You can extend ValkeyDict by adding or overriding encoding and decoding functions. This functionality enables various use cases, such as managing encrypted data in Valkey, To implement this, simply create and register your custom encoding and decoding functions. By delegating serialization to valkey-dict, reduce complexity and have simple code in the codebase.

__init__(namespace='main', expire=None, preserve_expiration=False, valkey=None, **valkey_kwargs)

Initialize a ValkeyDict instance.

Init the ValkeyDict instance.

Parameters:

• namespace (str) – A prefix for keys stored in Valkey.

• expire (Union[int, timedelta, None], optional) – Expiration time for keys.

• preserve_expiration (Optional[bool], optional) – Preserve expiration on key updates.

• valkey (Optional[StrictValkey[Any]], optional) – A Valkey connection instance.

• **valkey_kwargs (Any) – Additional kwargs for Valkey connection if not provided.

clear()

Remove all key-value pairs from the ValkeyDict in one batch operation using pipelining.

This method mimics the behavior of the clear method from a standard Python dictionary. Valkey pipelining is employed to group multiple commands into a single request, minimizing network round-trip time, latency, and I/O load, thereby enhancing the overall performance.

Return type:

None

multi_chain_get(_keys)

Not part of Python Valkey Dict.

Parameters:

_keys (List[str]) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

List[Any]

multi_del(_key)

Not part of Python Valkey Dict.

Parameters:

_key (str) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

int

multi_dict(_key)

Not part of Python Valkey Dict.

Parameters:

_key (str) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

Dict[str, Any]

multi_get(_key)

Not part of Python Valkey Dict.

Parameters:

_key (str) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

List[Any]

popitem()

Remove and return a random (key, value) pair from the ValkeyDict as a tuple.

This method is analogous to the popitem method of a standard Python dictionary.

if dict_compliant set true stays true to In Python 3.7+, removes the last inserted item (LIFO order)

Returns:

A tuple containing a randomly chosen (key, value) pair.

Return type:

tuple

Raises:

KeyError – If ValkeyDict is empty.

setdefault(key, default_value=None)

Get value under key, and if not present set default value.

Return the value associated with the given key if it exists, otherwise set the value to the default value and return it. Analogous to a dictionary’s setdefault method.

Parameters:

• key (str) – The key to retrieve the value.

• default_value (Optional[Any], optional) – The value to set if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

valkey_dict.type_management module

Type management module.

class valkey_dict.type_management.ValkeyDictJSONDecoder(*args, **kwargs)

Bases: JSONDecoder

JSON decoder leveraging ValkeyDict existing type conversion system.

Works with ValkeyDictJSONEncoder to reconstruct Python objects from JSON using ValkeyDict decoding_registry.

Still needs work but allows for more types than without.

__init__(*args, **kwargs)

Overwrite the __init__ method from JSON decoder.

Parameters:

• *args (Any) – Positional arguments for initialization.

• **kwargs (Any) – Keyword arguments for initialization.

class valkey_dict.type_management.ValkeyDictJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

Extends JSON encoding capabilities by reusing ValkeyDict type conversion.

Uses existing decoding_registry to know which types to handle specially and encoding_registry (falls back to str) for converting to JSON-compatible formats.

Example

The encoded format looks like:

{
    "__type__": "TypeName",
    "value": <encoded value>
}

Notes

Uses decoding_registry (containing all supported types) to check if type needs special handling. For encoding, defaults to str() if no encoder exists in encoding_registry.

default(o)

Overwrite default from json encoder.

Parameters:

o (Any) – Object to be serialized.

Raises:

TypeError – If the object o cannot be serialized.

Returns:

Serialized value.

Return type:

Any

valkey_dict.type_management.decode_json(s)

Decode a JSON string to a Python object using the existing decoding registry.

Parameters:

s (str) – The JSON string to be decoded.

Returns:

The decoded Python object.

Return type:

Any

valkey_dict.type_management.encode_json(obj)

Encode a Python object to a JSON string using the existing encoding registry.

Parameters:

obj (Any) – The Python object to be encoded.

Returns:

The JSON-encoded string representation of the object.

Return type:

str

Module contents

__init__ module for valkey dict.

class valkey_dict.PythonValkeyDict(namespace='main', expire=None, preserve_expiration=False, valkey=None, **valkey_kwargs)

Bases: ValkeyDict

Python dictionary with Valkey as backend.

With support for advanced features, such as custom data types, pipelining, and key expiration.

This class focuses on having one-to-on behavior of a dictionary while using Valkey as storage layer, allowing for efficient storage and retrieval of key-value pairs. It supports various data types, including strings, integers, floats, lists, dictionaries, tuples, sets, and user-defined types. The class leverages the power of Valkey pipelining to minimize network round-trip time, latency, and I/O load, thereby optimizing performance for batch operations. Additionally, it allows for the management of key expiration through the use of context managers.

The ValkeyDict class is designed to be analogous to a standard Python dictionary while providing enhanced functionality, such as support for a wider range of data types and efficient batch operations. It aims to offer a seamless and familiar interface for developers familiar with Python dictionaries, enabling a smooth transition to a Valkey-backed data store.

Extendable Types: You can extend ValkeyDict by adding or overriding encoding and decoding functions. This functionality enables various use cases, such as managing encrypted data in Valkey, To implement this, simply create and register your custom encoding and decoding functions. By delegating serialization to valkey-dict, reduce complexity and have simple code in the codebase.

__init__(namespace='main', expire=None, preserve_expiration=False, valkey=None, **valkey_kwargs)

Initialize a ValkeyDict instance.

Init the ValkeyDict instance.

Parameters:

• namespace (str) – A prefix for keys stored in Valkey.

• expire (Union[int, timedelta, None], optional) – Expiration time for keys.

• preserve_expiration (Optional[bool], optional) – Preserve expiration on key updates.

• valkey (Optional[StrictValkey[Any]], optional) – A Valkey connection instance.

• **valkey_kwargs (Any) – Additional kwargs for Valkey connection if not provided.

clear()

Remove all key-value pairs from the ValkeyDict in one batch operation using pipelining.

This method mimics the behavior of the clear method from a standard Python dictionary. Valkey pipelining is employed to group multiple commands into a single request, minimizing network round-trip time, latency, and I/O load, thereby enhancing the overall performance.

Return type:

None

multi_chain_get(_keys)

Not part of Python Valkey Dict.

Parameters:

_keys (List[str]) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

List[Any]

multi_del(_key)

Not part of Python Valkey Dict.

Parameters:

_key (str) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

int

multi_dict(_key)

Not part of Python Valkey Dict.

Parameters:

_key (str) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

Dict[str, Any]

multi_get(_key)

Not part of Python Valkey Dict.

Parameters:

_key (str) – Not used.

Raises:

NotImplementedError – Not part of Python Valkey Dict.

Return type:

List[Any]

popitem()

Remove and return a random (key, value) pair from the ValkeyDict as a tuple.

This method is analogous to the popitem method of a standard Python dictionary.

if dict_compliant set true stays true to In Python 3.7+, removes the last inserted item (LIFO order)

Returns:

A tuple containing a randomly chosen (key, value) pair.

Return type:

tuple

Raises:

KeyError – If ValkeyDict is empty.

setdefault(key, default_value=None)

Get value under key, and if not present set default value.

Return the value associated with the given key if it exists, otherwise set the value to the default value and return it. Analogous to a dictionary’s setdefault method.

Parameters:

• key (str) – The key to retrieve the value.

• default_value (Optional[Any], optional) – The value to set if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

class valkey_dict.ValkeyDict(namespace='main', expire=None, preserve_expiration=False, valkey=None, raise_key_error_delete=False, **valkey_kwargs)

Bases: object

Python dictionary with Valkey as backend.

With support for advanced features, such as custom data types, pipelining, and key expiration.

This class provides a dictionary-like interface that interacts with a Valkey database, allowing for efficient storage and retrieval of key-value pairs. It supports various data types, including strings, integers, floats, lists, dictionaries, tuples, sets, and user-defined types. The class leverages the power of Valkey pipelining to minimize network round-trip time, latency, and I/O load, thereby optimizing performance for batch operations. Additionally, it allows for the management of key expiration through the use of context managers.

The ValkeyDict class is designed to be analogous to a standard Python dictionary while providing enhanced functionality, such as support for a wider range of data types and efficient batch operations. It aims to offer a seamless and familiar interface for developers familiar with Python dictionaries, enabling a smooth transition to a Valkey-backed data store.

Extendable Types: You can extend ValkeyDict by adding or overriding encoding and decoding functions. This functionality enables various use cases, such as managing encrypted data in Valkey, To implement this, simply create and register your custom encoding and decoding functions. By delegating serialization to valkey-dict, reduce complexity and have simple code in the codebase.

__init__(namespace='main', expire=None, preserve_expiration=False, valkey=None, raise_key_error_delete=False, **valkey_kwargs)

Initialize a ValkeyDict instance.

Init the ValkeyDict instance.

Parameters:

• namespace (str) – A prefix for keys stored in Valkey.

• expire (Union[int, timedelta, None], optional) – Expiration time for keys.

• preserve_expiration (Optional[bool], optional) – Preserve expiration on key updates.

• valkey (Optional[StrictValkey[Any]], optional) – A Valkey connection instance.

• raise_key_error_delete (bool) – Enable strict Python dict behavior raise if key not found when deleting.

• **valkey_kwargs (Any) – Additional kwargs for Valkey connection if not provided.

chain_del(iterable)

Delete a value from the ValkeyDict using a chain of keys.

Parameters:

iterable (List[str]) – A list of keys representing the chain.

Return type:

None

chain_get(iterable)

Get a value from the ValkeyDict using a chain of keys.

Parameters:

iterable (List[str]) – A list of keys representing the chain.

Returns:

The value associated with the chain of keys.

Return type:

Any

chain_set(iterable, v)

Set a value in the ValkeyDict using a chain of keys.

Parameters:

• iterable (List[str]) – A list of keys representing the chain.

• v (Any) – The value to be set.

Return type:

None

clear()

Remove all key-value pairs from the ValkeyDict in one batch operation using pipelining.

This method mimics the behavior of the clear method from a standard Python dictionary. Valkey pipelining is employed to group multiple commands into a single request, minimizing network round-trip time, latency, and I/O load, thereby enhancing the overall performance.

Return type:

None

copy()

Create a shallow copy of the ValkeyDict and return it as a standard Python dictionary.

This method is analogous to the copy method of a standard Python dictionary

Returns:

A shallow copy of the ValkeyDict as a standard Python dictionary.

Return type:

dict

Note

does not create a new ValkeyDict instance.

decoding_registry: Dict[str, Callable[[str], Any]] = {'Decimal': <class 'decimal.Decimal'>, 'NoneType': <function <lambda>>, 'OrderedDict': <function <lambda>>, 'UUID': <class 'uuid.UUID'>, 'bool': <function <lambda>>, 'bytes': <function b64decode>, 'complex': <function <lambda>>, 'date': <built-in method fromisoformat of type object>, 'datetime': <built-in method fromisoformat of type object>, 'defaultdict': <function <lambda>>, 'dict': <function decode_json>, 'float': <class 'float'>, 'frozenset': <function <lambda>>, 'int': <class 'int'>, 'list': <function decode_json>, 'set': <function _decode_set>, 'str': <class 'str'>, 'time': <built-in method fromisoformat of type object>, 'timedelta': <function <lambda>>, 'tuple': <function _decode_tuple>}

encoding_registry: Dict[str, Callable[[Any], str]] = {'OrderedDict': <function <lambda>>, 'bytes': <function <lambda>>, 'complex': <function <lambda>>, 'date': <method 'isoformat' of 'datetime.date' objects>, 'datetime': <method 'isoformat' of 'datetime.datetime' objects>, 'defaultdict': <function <lambda>>, 'dict': <function encode_json>, 'frozenset': <function <lambda>>, 'list': <function encode_json>, 'set': <function _encode_set>, 'time': <method 'isoformat' of 'datetime.time' objects>, 'timedelta': <function <lambda>>, 'tuple': <function _encode_tuple>}

expire: Union[int, timedelta, None]

expire_at(sec_epoch)

Context manager to set the expiration time for keys in the ValkeyDict.

Parameters:

sec_epoch (int, timedelta) – The expiration duration is set using either an integer or a timedelta.

Yields:

ContextManager – A context manager during which the expiration time is the time set.

Return type:

Iterator[None]

extends_type(class_type, encode=None, decode=None, encoding_method_name=None, decoding_method_name=None)

Extend ValkeyDict to support a custom type in the encode/decode mapping.

This method enables serialization of instances based on their type, allowing for custom types, specialized storage formats, and more. There are three ways to add custom types: 1. Have a class with an encode instance method and a decode class method. 2. Have a class and pass encoding and decoding functions, where encode converts the class instance to a string, and decode takes the string and recreates the class instance. 3. Have a class that already has serialization methods, that satisfies the: EncodeFuncType = Callable[[Any], str] DecodeFuncType = Callable[[str], Any]

custom_encode_method custom_decode_method

If no encoding or decoding function is provided, default to use the encode and decode methods of the class.

The encode method should be an instance method that converts the object to a string. The decode method should be a class method that takes a string and returns an instance of the class.

The method names for encoding and decoding can be changed by modifying the - custom_encode_method - custom_decode_method attributes of the ValkeyDict instance

Example

>>> class Person:
...     def __init__(self, name, age):
...         self.name = name
...         self.age = age
...
...     def encode(self) -> str:
...         return json.dumps(self.__dict__)
...
...     @classmethod
...     def decode(cls, encoded_str: str) -> 'Person':
...         return cls(**json.loads(encoded_str))
...
>>> valkey_dict.extends_type(Person)

Parameters:

• class_type (type) – The class __name__ will become the key for the encoding and decoding functions.

• encode (Optional[EncodeFuncType]) – function that encodes an object into a storable string format.

• decode (Optional[DecodeFuncType]) – function that decodes a string back into an object of class_type.

• encoding_method_name (str, optional) – Name of encoding method of the class for valkey-dict custom types.

• decoding_method_name (str, optional) – Name of decoding method of the class for valkey-dict custom types.

Raises:

NotImplementedError –

Return type:

None

Note

You can check for compliance of a class separately using the new_type_compliance method:

This method raises a NotImplementedError if either encode or decode is None and the class does not implement the corresponding method.

fromkeys(iterable, value=None)

Create a new ValkeyDict from an iterable of key-value pairs.

Create a new ValkeyDict with keys from the provided iterable and values set to the given value. This method is analogous to the fromkeys method of a standard Python dictionary, populating the ValkeyDict with the keys from the iterable and setting their corresponding values to the specified value.

Parameters:

• iterable (List[str]) – An iterable containing the keys to be added to the ValkeyDict.

• value (Optional[Any], optional) – The value to be assigned to each key in the ValkeyDict. Defaults to None.

Returns:

The current ValkeyDict instance,populated with the keys from the iterable and their corresponding values.

Return type:

ValkeyDict

get(key, default=None)

Return the value for the given key if it exists, otherwise return the default value.

Analogous to a dictionary’s get method.

Parameters:

• key (str) – The key to retrieve the value.

• default (Optional[Any], optional) – The value to return if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

get_ttl(key)

Get the Time To Live from Valkey.

Get the Time To Live (TTL) in seconds for a given key. If the key does not exist or does not have an associated expire, return None.

Parameters:

key (str) – The key for which to get the TTL.

Returns:

The TTL in seconds if the key exists and has an expiry set; otherwise, None.

Return type:

Optional[int]

get_valkey: StrictRedis[Any]

get_valkey_info()

Retrieve information and statistics about the Valkey server.

Returns:

The information and statistics from the Valkey server in a dictionary.

Return type:

dict

items()

Return a list of key-value pairs (tuples) in the ValkeyDict, analogous to a dictionary’s items method.

Yields:

Iterator[Tuple[str, Any]] – A list of key-value pairs in the ValkeyDict.

Return type:

Iterator[Tuple[str, Any]]

key(search_term='')

Return the first value for search_term if it exists, otherwise return None.

Parameters:

search_term (str) – A search term to filter keys. Defaults to ‘’.

Returns:

The first key associated with the given search term.

Return type:

str

keys()

Return an Iterator of keys in the ValkeyDict, analogous to a dictionary’s keys method.

Returns:

A list of keys in the ValkeyDict.

Return type:

Iterator[str]

multi_chain_get(keys)

Get multiple values from the ValkeyDict using a chain of keys.

Parameters:

keys (List[str]) – A list of keys representing the chain.

Returns:

A list of values associated with the chain of keys.

Return type:

List[Any]

multi_del(key)

Delete multiple values from the ValkeyDict using a shared key prefix.

Parameters:

key (str) – The shared key prefix.

Returns:

The number of keys deleted.

Return type:

int

multi_dict(key)

Get a dictionary of key-value pairs from the ValkeyDict using a shared key prefix.

Parameters:

key (str) – The shared key prefix.

Returns:

A dictionary of key-value pairs associated with the key prefix.

Return type:

Dict[str, Any]

multi_get(key)

Get multiple values from the ValkeyDict using a shared key prefix.

Parameters:

key (str) – The shared key prefix.

Returns:

A list of values associated with the key prefix.

Return type:

List[Any]

namespace: str

new_type_compliance(class_type, encode_method_name=None, decode_method_name=None)

Check if a class complies with the required encoding and decoding methods.

Parameters:

• class_type (type) – The class to check for compliance.

• encode_method_name (str, optional) – Name of encoding method of the class for valkey-dict custom types.

• decode_method_name (str, optional) – Name of decoding method of the class for valkey-dict custom types.

Raises:

NotImplementedError – If the class does not implement the required methods when the respective check is True.

Return type:

None

next()

Get the next item in the iterator (alias for __next__).

Returns:

The next item in the iterator.

Return type:

str

pipeline()

Context manager to create a Valkey pipeline for batch operations.

Yields:

ContextManager – A context manager to create a Valkey pipeline batching all operations within the context.

Return type:

Iterator[None]

pop(key, default=<object object>)

Analogous to a dictionary’s pop method.

Remove the value associated with the given key and return it, or return the default value if the key is not found.

Parameters:

• key (str) – The key to remove the value.

• default (Optional[Any], optional) – The value to return if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

Raises:

KeyError – If the key is not found and no default value is provided.

popitem()

Remove and return a random (key, value) pair from the ValkeyDict as a tuple.

This method is analogous to the popitem method of a standard Python dictionary.

if dict_compliant set true stays true to In Python 3.7+, removes the last inserted item (LIFO order)

Returns:

A tuple containing a randomly chosen (key, value) pair.

Return type:

tuple

Raises:

KeyError – If ValkeyDict is empty.

preserve_expiration: Optional[bool]

raise_key_error_delete: bool

setdefault(key, default_value=None)

Get value under key, and if not present set default value.

Return the value associated with the given key if it exists, otherwise set the value to the default value and return it. Analogous to a dictionary’s setdefault method.

Parameters:

• key (str) – The key to retrieve the value.

• default_value (Optional[Any], optional) – The value to set if the key is not found.

Returns:

The value associated with the key or the default value.

Return type:

Any

to_dict()

Convert the ValkeyDict to a Python dictionary.

Returns:

A dictionary with the same key-value pairs as the ValkeyDict.

Return type:

Dict[str, Any]

update(dic)

Update the ValkeyDict with key-value pairs from the given mapping, analogous to a dictionary’s update method.

Parameters:

dic (Mapping[str, Any]) – A mapping containing key-value pairs to update the ValkeyDict.

Return type:

None

valkey: StrictRedis[Any]

values()

Analogous to a dictionary’s values method.

Return a list of values in the ValkeyDict,

Yields:

List[Any] – A list of values in the ValkeyDict.

Return type:

Iterator[Any]

class valkey_dict.ValkeyDictJSONDecoder(*args, **kwargs)

Bases: JSONDecoder

JSON decoder leveraging ValkeyDict existing type conversion system.

Works with ValkeyDictJSONEncoder to reconstruct Python objects from JSON using ValkeyDict decoding_registry.

Still needs work but allows for more types than without.

__init__(*args, **kwargs)

Overwrite the __init__ method from JSON decoder.

Parameters:

• *args (Any) – Positional arguments for initialization.

• **kwargs (Any) – Keyword arguments for initialization.

class valkey_dict.ValkeyDictJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

Extends JSON encoding capabilities by reusing ValkeyDict type conversion.

Uses existing decoding_registry to know which types to handle specially and encoding_registry (falls back to str) for converting to JSON-compatible formats.

Example

The encoded format looks like:

{
    "__type__": "TypeName",
    "value": <encoded value>
}

Notes

Uses decoding_registry (containing all supported types) to check if type needs special handling. For encoding, defaults to str() if no encoder exists in encoding_registry.

default(o)

Overwrite default from json encoder.

Parameters:

o (Any) – Object to be serialized.

Raises:

TypeError – If the object o cannot be serialized.

Returns:

Serialized value.

Return type:

Any