Skip to content

Instantly share code, notes, and snippets.

@dln
Created August 14, 2010 11:32
Show Gist options
  • Save dln/524216 to your computer and use it in GitHub Desktop.
Save dln/524216 to your computer and use it in GitHub Desktop.
Implementaion a dictionary with timed expiration of values.
import time
class TtlDict(dict):
"""A dict where values expire after a set time.
Expiration is done on access, or using `prune` (linear)"""
ttl = 5
def __init__(self, *args, **kwargs):
dict.__init__(self, *args, **kwargs)
expires = time.time() + self.ttl
self._expirations = dict((k, expires) for k in dict.keys(self))
def __setitem__(self, key, val):
dict.__setitem__(self, key, val)
self._expirations[key] = time.time() + self.ttl
def __getitem__(self, key):
expires = self._expirations.get(key, None)
if expires and expires < time.time():
del self[key]
return dict.__getitem__(self, key)
def __delitem__(self, key):
dict.__delitem__(self, key)
try:
del self._expirations[key]
except KeyError:
pass
def iterkeys(self, delete_expired=False):
"""Iterate over keys, optionally deleting expired items upon completion."""
t = time.time()
expired_keys = []
try:
for key, expires in self._expirations.iteritems():
if expires >= t:
yield key
elif delete_expired:
expired_keys.append(key)
finally:
for key in expired_keys:
del self[key]
__iter__ = iterkeys
def iteritems(self, delete_expired=False):
"""Iterate over keys, optionally deleting expired items upon completion."""
for key in self.iterkeys(delete_expired):
yield key, dict.__getitem__(self, key)
def keys(self):
"""Return list of active keys. Deletes expired entries."""
fresh_keys = []
t = time.time()
for key, expires in self._expirations.items():
if expires >= t:
fresh_keys.append(key)
else:
del self[key]
return fresh_keys
def items(self):
"""Return list of active items. Deletes expired entries."""
fresh_items = []
t = time.time()
for key, expires in self._expirations.items():
if expires >= t:
fresh_items.append((key, dict.__getitem__(self, key)))
else:
del self[key]
return fresh_items
def update(self, E, **F):
for key in E:
self[key] = E[key]
if F:
self.update(F)
def clear(self):
dict.clear(self)
self._expirations.clear()
def prune(self, expire_time=None):
"""Delete all keys expired before `expire_time`. Default is now."""
t = time.time() if expire_time is None else expire_time
for key, expires in self._expirations.items():
if expires < t:
del self[key]
def touch(self, key, ts=None, ttl=ttl):
"""Change lifetime for a key, either to an absolute expiration time `ts`, or relative to now `ttl` """
self._expirations[key] = (time.time() + ttl) if ts is None else ts
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment