Created
August 14, 2010 11:32
-
-
Save dln/524216 to your computer and use it in GitHub Desktop.
Implementaion a dictionary with timed expiration of values.
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
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