-
Notifications
You must be signed in to change notification settings - Fork 7
/
cache.py
67 lines (50 loc) · 1.92 KB
/
cache.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
import os
from urllib.parse import urlparse
import redis
DEFAULT_EXPIRES = 60 * 15 # number of seconds before we auto-expire keys
KEY_PREFIX = "{{APP_SLUG}}"
u = urlparse(os.environ["REDIS_URL"])
redis_connection = redis.StrictRedis(host=u.hostname, port=u.port, password=u.password, db=0)
class RedisCache(object):
"""docstring for RedisCache"""
def __init__(self):
super(RedisCache, self).__init__()
if not os.environ["REDIS_URL"]:
raise Exception("Can't use redis without specifying valid `REDIS_URL` env variable.")
self.connection = redis_connection
def _build_key(self, k):
if type(k) == bytes:
k = k.decode() # convert bytes to str
if k.startswith(f"{KEY_PREFIX}:"):
return k
return f"{KEY_PREFIX}:{k}"
def set(self, k, v, expires=DEFAULT_EXPIRES):
k = self._build_key(k)
return self.connection.set(k, v, ex=expires)
def get(self, k, default=None):
k = self._build_key(k)
return self.connection.get(k) or default
def set_dict(self, k, v):
return self.connection.hmset(k, v)
def get_dict(self, k, default=None):
data = self.connection.hgetall(k)
return { key.decode(): val.decode() for key, val in data.items() }
def delete(self, k):
k = self._build_key(k)
return self.connection.delete(k)
def scan_iter(self, k):
k = self._build_key(k)
return self.connection.scan_iter(k)
def delete_pattern(self, pattern):
for key in self.scan_iter(pattern):
self.delete(key)
# instantiate it so we can just import the name 'redis' and use it anywhere
redis = RedisCache()
if __name__ == '__main__':
# test connection
redis.set('test:foo', 'bar')
redis.set('test:hello', 'world')
for k in redis.scan_iter('test:*'):
v = redis.get(k)
print(f"{k} => {v}")
redis.delete_pattern("test:*")