-
Notifications
You must be signed in to change notification settings - Fork 0
/
class_redis.py
78 lines (68 loc) · 2.51 KB
/
class_redis.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import redis
import os
import sys
import json
sys.path.insert(0, '/Users/Joen/Documents/COS-CM-BOT/secret')
try:
print("using environment variables")
REDIS_URL = os.environ['REDIS_URL']
except:
import secret.keys as keys
print("moving on to redis secrets")
REDIS_URL = keys.REDIS_KEY
#handles redis db, navigates and edits values
class RedisHandler:
def __init__(self):
self.r = redis.from_url(REDIS_URL)
def split_paths(self, path):
dirs = path.split('/')
main_key = dirs[0]
subsequent_dirs = dirs[1:]
return main_key, subsequent_dirs
# RECURSIVE NAVIGATION
#----------------------
#recursively navigates to object value via paths and edits value
def add_value_from_path(self, paths:list, obj, value):
if len(paths) > 1:
new_obj = obj[paths[0]]
paths.pop(0)
self.add_value_from_path(paths, new_obj, value)
elif len(paths) == 1:
obj[paths[0]] = value
return obj
#recursively navigates to object value via paths and returns value
def navigate_to_value_from_path(self, paths:list, obj):
if len(paths) > 1:
new_obj = obj[paths[0]]
paths.pop(0)
self.navigate_to_value_from_path(paths, new_obj)
elif len(paths) == 1:
return obj[paths[0]]
#----------------------
#changes value at endpoint of paths
def edit_value_from_path(self, paths:str, value):
main_key, subsequent_dirs = self.split_paths(paths)
if len(subsequent_dirs) > 0:
obj = json.loads(self.r.get(main_key))
edited_obj = self.add_value_from_path(subsequent_dirs, obj, value)
self.r.set(main_key, json.dumps(edited_obj))
else:
value_type = type(value)
if value_type == str:
self.r.set(main_key, value)
else:
self.r.set(main_key, json.dumps(value))
#reads value at endpoint of paths
def read_value_from_path(self, paths:str):
main_key, subsequent_dirs = self.split_paths(paths)
if len(subsequent_dirs) > 0:
obj = json.loads(self.r.get(main_key))
return self.navigate_to_value_from_path(subsequent_dirs, obj)
else:
return self.r.get(main_key)
#db = RedisHandler(REDIS_URL)
#print(db.edit_value_from_path("hello",0))
#print(db.read_value_from_path("hello/hi"))
#if __name__ == "__main__":
# db = RedisHandler()
# db.r.set("test_exp", "testing this", ex=60)