redis_dict.core module

Redis Dict module.

class redis_dict.core.RedisDict(namespace='main', expire=None, preserve_expiration=False, redis=None, raise_key_error_delete=False, **redis_kwargs)

Bases: object

Python dictionary with Redis 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 Redis 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 Redis 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 RedisDict 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 Redis-backed data store.

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

__init__(namespace='main', expire=None, preserve_expiration=False, redis=None, raise_key_error_delete=False, **redis_kwargs)

Initialize a RedisDict instance.

Init the RedisDict instance.

Parameters:

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

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

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

• redis (Optional[StrictRedis[Any]], optional) – A Redis connection instance.

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

• **redis_kwargs (Any) – Additional kwargs for Redis connection if not provided.

chain_del(iterable)

Delete a value from the RedisDict 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 RedisDict 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 RedisDict 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 RedisDict in one batch operation using pipelining.

This method mimics the behavior of the clear method from a standard Python dictionary. Redis 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 RedisDict 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 RedisDict as a standard Python dictionary.

Return type:

dict

Note

does not create a new RedisDict 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 RedisDict.

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 RedisDict 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 RedisDict 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))
...
>>> redis_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 redis-dict custom types.

• decoding_method_name (str, optional) – Name of decoding method of the class for redis-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 RedisDict from an iterable of key-value pairs.

Create a new RedisDict 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 RedisDict 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 RedisDict.

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

Returns:

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

Return type:

RedisDict

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_redis: StrictRedis[Any]

get_redis_info()

Retrieve information and statistics about the Redis server.

Returns:

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

Return type:

dict

get_ttl(key)

Get the Time To Live from Redis.

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]

items()

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

Yields:

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

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 RedisDict, analogous to a dictionary’s keys method.

Returns:

A list of keys in the RedisDict.

Return type:

Iterator[str]

multi_chain_get(keys)

Get multiple values from the RedisDict 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 RedisDict 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 RedisDict 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 RedisDict 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 redis-dict custom types.

• decode_method_name (str, optional) – Name of decoding method of the class for redis-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 Redis pipeline for batch operations.

Yields:

ContextManager – A context manager to create a Redis 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 RedisDict 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 RedisDict is empty.

preserve_expiration: Optional[bool]

raise_key_error_delete: bool

redis: StrictRedis[Any]

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 RedisDict to a Python dictionary.

Returns:

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

Return type:

Dict[str, Any]

update(dic)

Update the RedisDict 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 RedisDict.

Return type:

None

values()

Analogous to a dictionary’s values method.

Return a list of values in the RedisDict,

Yields:

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

Return type:

Iterator[Any]