Skip to content

Instantly share code, notes, and snippets.

@j105rob
Created May 23, 2015 11:01
Show Gist options
  • Save j105rob/66a41f050b29c903336f to your computer and use it in GitHub Desktop.
Save j105rob/66a41f050b29c903336f to your computer and use it in GitHub Desktop.
Update to mailgun's ExpiringDict to add a callback func when expiration happens.
class ExpiringDictWithCallback(ExpiringDict):
def __init__(self, max_len, max_age_seconds, callback=None):
assert max_age_seconds >= 0
assert max_len >= 1
OrderedDict.__init__(self)
self.max_len = max_len
self.max_age = max_age_seconds
self.lock = RLock()
self.callback = callback
def __contains__(self, key):
""" Return True if the dict has a key, else return False. """
try:
with self.lock:
item = OrderedDict.__getitem__(self, key)
if time.time() - item[1] < self.max_age:
return True
else:
if self.callback:self.callback(item[1])
del self[key]
except KeyError:
pass
return False
def __getitem__(self, key, with_age=False):
""" Return the item of the dict.
Raises a KeyError if key is not in the map.
"""
with self.lock:
item = OrderedDict.__getitem__(self, key)
item_age = time.time() - item[1]
if item_age < self.max_age:
if with_age:
return item[0], item_age
else:
return item[0]
else:
if self.callback:self.callback(item[1])
del self[key]
raise KeyError(key)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment