Created
February 20, 2018 23:29
-
-
Save D-Nice/0b48dd9833e9189fb6bc15f9a842e198 to your computer and use it in GitHub Desktop.
md5 hash memory dict cache
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from datetime import datetime, timedelta | |
from hashlib import md5 | |
from copy import deepcopy | |
import json | |
# creates an md5 hash dict for caching in memory | |
class Cache: | |
# args expect seconds | |
def __init__(self, expiry, clearInterval): | |
if (not type(expiry) == int) or (not type(clearInterval) == int): | |
raise TypeError('Expected int type') | |
self.dict = {} | |
self.expiry = expiry | |
self.clearInterval = clearInterval | |
self.lastCleared = datetime.now() | |
def __setitem__(self, key, value, hashed=False): | |
timestamp = datetime.now() | |
key = key if hashed else self._hashedKey(key) | |
self.dict[key] = { 'val': value, 'ts': timestamp } | |
def __delitem__(self, key, hashed=False): | |
key = key if hashed else self._hashedKey(key) | |
del self.dict[key] | |
def __getitem__(self, key, hashed=False): | |
key = key if hashed else self._hashedKey(key) | |
return self.dict[key]['val'] | |
def _clearOldCache(self): | |
elapsedSecs = (datetime.now() - self.lastCleared).seconds | |
if elapsedSecs > self.clearInterval: | |
temp = deepcopy(self.dict) | |
for key in temp: | |
if not self.isCached(key, True): | |
del self.dict[key] | |
del temp | |
def getTime(self, key, hashed): | |
key = key if hashed else self._hashedKey(key) | |
return self.dict[key]['ts'] | |
def getFull(self, key, hashed=False): | |
key = key if hashed else self._hashedKey(key) | |
return self.dict[key] | |
def _hashedKey(self, key): | |
h = md5() | |
s = json.dumps(key).encode('utf-8') | |
h.update(s) | |
return h.hexdigest() | |
def isCached(self, key, hashed=False): | |
key = key if hashed else self._hashedKey(key) | |
try: | |
self.__getitem__(key, True) | |
except KeyError: | |
return False | |
t0 = self.getTime(key, True) | |
elapsedSecs = (datetime.now() - t0).seconds | |
if elapsedSecs > self.expiry: | |
return False | |
return True | |
yo = Cache('3', 4) | |
#yo.set('5', '10') | |
j = json.loads('{ "test": 1 }') | |
yo[j] = '10' | |
print(yo[j]) | |
yo._clearOldCache() | |
time.sleep(5) | |
print(yo.isCached(j)) | |
yo._clearOldCache() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment