import json import redis class RedisDict: def __init__(self, name, redis_url): self.name = name self.redis = redis.Redis.from_url(redis_url, decode_responses=True) def __setitem__(self, key, value): serialized_value = json.dumps(value) self.redis.hset(self.name, key, serialized_value) def __getitem__(self, key): value = self.redis.hget(self.name, key) if value is None: raise KeyError(key) return json.loads(value) def __delitem__(self, key): result = self.redis.hdel(self.name, key) if result == 0: raise KeyError(key) def __contains__(self, key): return self.redis.hexists(self.name, key) def __len__(self): return self.redis.hlen(self.name) def keys(self): return self.redis.hkeys(self.name) def values(self): return [json.loads(v) for v in self.redis.hvals(self.name)] def items(self): return [(k, json.loads(v)) for k, v in self.redis.hgetall(self.name).items()] def get(self, key, default=None): try: return self[key] except KeyError: return default def clear(self): self.redis.delete(self.name) def update(self, other=None, **kwargs): if other is not None: for k, v in other.items() if hasattr(other, "items") else other: self[k] = v for k, v in kwargs.items(): self[k] = v def setdefault(self, key, default=None): if key not in self: self[key] = default return self[key]