Created
June 8, 2017 06:56
-
-
Save s8sg/a2f77ed2fe46d734f9b85116f444c17b to your computer and use it in GitHub Desktop.
A cache dict implementation with object level timeout
This file contains hidden or 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
import time | |
import collections | |
class Cache(collections.MutableMapping): | |
def __init__(self, timeout=1, *args, **kwargs): | |
self.__store = dict() | |
self.__time_table = dict() | |
self.__time_out = timeout | |
# use the free update to set keys | |
self.update(dict(*args, **kwargs)) | |
def __getitem__(self, key): | |
if time.time() >= self.__time_table.get(key): | |
del self.__store[key] | |
del self.__time_table[key] | |
return self.__store[key] | |
def __setitem__(self, key, value): | |
self.__store[key] = value | |
self.__time_table[key] = time.time() + self.__time_out | |
def __delitem__(self, key): | |
del self.__store[key] | |
del self.__time_table[key] | |
def __contains__(self, item): | |
if self.__time_table.get(item, None) is None: | |
return False | |
if time.time() >= self.__time_table.get(item): | |
del self.__store[item] | |
del self.__time_table[item] | |
return False | |
return True | |
def __iter__(self): | |
for item in iter(self.__store): | |
if time.time() < self.__time_table.get(item): | |
yield item | |
def __len__(self): | |
for item in iter(self.__store): | |
if time.time() >= self.__time_table.get(item): | |
del self.__store[item] | |
del self.__time_table[item] | |
return len(self.__store) | |
''' | |
cache = Cache(3) | |
cache["k1"] = "data1" | |
cache["k2"] = "data2" | |
time.sleep(1) | |
cache["k3"] = "data3" | |
time.sleep(1) | |
cache["k4"] = "data4" | |
while True: | |
for item in iter(cache): | |
print(item, cache[item]) | |
time.sleep(1) | |
''' |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment