An interface to redis HSET, wrapped in a dictionary, using django_redis.

dictionary_cache.py

from collections.abc import MutableMapping

from django_redis.cache import RedisCache
from redis.client import Redis


class DictionaryCache(MutableMapping):

    def __init__(
        self,
        key: str,
        *,
        cache: RedisCache,
        key_cast=str,
        value_cast=str,
    ):
        self._name = key
        self._redis: Redis = cache.client.get_client()
        self._key_cast = key_cast
        self._value_cast = value_cast

    def __len__(self):
        return self._redis.hlen(self._name)

    def __getitem__(self, key):
        if key not in self:
            raise KeyError(key)
        return self._redis_to_value(self._redis.hget(self._name, self._key_to_redis(key)))

    def __setitem__(self, key, item):
        self._redis.hset(self._name, self._key_to_redis(key), self._value_to_redis(item))

    def __delitem__(self, key):
        self._redis.hdel(self._name, self._key_to_redis(key))

    def __iter__(self):
        return iter(self.keys())

    def __contains__(self, key):
        return self._redis.hexists(self._name, self._key_to_redis(key))

    def __repr__(self):
        return repr(self.as_dict())

    def __or__(self, other):
        return NotImplemented

    def __ror__(self, other):
        return NotImplemented

    def __ior__(self, other):
        raise NotImplemented

    def __copy__(self):
        raise NotImplemented

    def update(self, *args, **kwargs):
        output = dict(*args, **kwargs)
        output = {self._key_to_redis(k): self._value_to_redis(v) for k, v in output.items()}
        self._redis.hset(self._name, mapping=output)

    def keys(self):
        return (self._redis_to_key(key) for key in self._redis.hkeys(self._name))

    def values(self):
        return (self._redis_to_value(key) for key in self._redis.hvals(self._name))

    def items(self):
        return self.as_dict().items()

    def clear(self):
        self._redis.delete(self._name)

    def as_dict(self):
        return {
            self._redis_to_key(k): self._redis_to_value(v)
            for k, v in self._redis.hgetall(self._name).items()
        }

    @staticmethod
    def _encode_str(data: str) -> bytes:
        return str(data).encode('utf-8')

    @staticmethod
    def _decode_str(data: bytes) -> str:
        return bytes(data).decode('utf-8')

    def _key_to_redis(self, key):
        return self._encode_str(key)

    def _redis_to_key(self, key):
        return self._key_cast(self._decode_str(key))

    def _value_to_redis(self, value):
        return self._encode_str(value)

    def _redis_to_value(self, value):
        return self._value_cast(self._decode_str(value))