Skip to content

Instantly share code, notes, and snippets.

@AmirSbss
Last active June 1, 2022 19:21
Show Gist options
  • Save AmirSbss/2644a59178d4b4e99bd959805ed0ff87 to your computer and use it in GitHub Desktop.
Save AmirSbss/2644a59178d4b4e99bd959805ed0ff87 to your computer and use it in GitHub Desktop.
Use redis document-oriented!
import string
import random
import pickle
from redis import StrictRedis
def random_id(lenght=8, letters=string.ascii_letters + string.digits):
return "".join(random.choice(letters) for _ in range(lenght))
class DB:
def __init__(self, client=StrictRedis()):
self.redis = client
def get_str(self, value):
if isinstance(value, str):
return value
elif isinstance(value, list):
strvalue = "list:" + random_id()
for i in value:
self.redis.sadd(strvalue, self.get_str(i))
return strvalue
elif isinstance(value, dict):
strvalue = "dict:" + random_id()
for i in value.keys():
self.redis.hset(strvalue, i, self.get_str(value[i]))
return strvalue
else:
strvalue = "pickle:" + random_id()
self.redis.set(strvalue, pickle.dumps(value))
return strvalue
def get_real(self, strvalue):
if strvalue.startswith("list:"):
data = []
for i in self.redis.smembers(strvalue):
data.append(self.get_real(i))
return data
elif strvalue.startswith("dict:"):
data = {}
for i in self.redis.hkeys(strvalue):
data[i] = self.get_real(self.redis.hget(strvalue, i))
return data
elif strvalue.startswith("pickle:"):
data = pickle.loads(self.redis.get(strvalue))
return data
else:
return strvalue
def del_real(self, strvalue):
if strvalue.startswith("list:"):
for i in self.redis.smembers(strvalue):
self.del_real(i)
elif strvalue.startswith("dict:"):
for i in self.redis.hkeys(strvalue):
self.del_real(self.redis.hget(strvalue, i))
return self.redis.delete(strvalue)
def set(self, name, value):
self.delete(name)
self.redis.set(name, self.get_str(value))
def get(self, name):
return self.get_real(self.redis.get(str(name)))
def delete(self, name):
return self.del_real(str(name))
def update(self, name, value):
names = name.split(".")
if len(names) == 1:
return self.set(name, value)
parent_hash = self.redis.get(names[0])
sub_hash = names[1]
i = 1
while i < len(names)-1:
parent_hash = self.redis.hget(parent_hash, sub_hash)
sub_hash = names[i + 1]
i += 1
strvalue = self.redis.hget(parent_hash, sub_hash)
self.delete(strvalue)
self.redis.hset(parent_hash, sub_hash, self.get_str(value))
db = DB()
a = {
"testing": True,
"working": "yes!",
"test_list": ["it's working!"],
"test_dict": {"working": "fine"},
"test_int": 1
}
db.set("test", a)
loaded = db.get("test")
print "loaded == a: ", cmp(a, loaded) == 0
db.update("test.working", "Hell yes!")
print "Updates version: ", db.get("test")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment