mirror of
https://github.com/open-webui/open-webui
synced 2024-11-18 06:22:56 +00:00
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
|
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]
|