|
import time |
|
from collections.abc import MutableMapping |
|
from functools import lru_cache |
|
|
|
|
|
class DirCache(MutableMapping): |
|
""" |
|
Caching of directory listings, in a structure like:: |
|
|
|
{"path0": [ |
|
{"name": "path0/file0", |
|
"size": 123, |
|
"type": "file", |
|
... |
|
}, |
|
{"name": "path0/file1", |
|
}, |
|
... |
|
], |
|
"path1": [...] |
|
} |
|
|
|
Parameters to this class control listing expiry or indeed turn |
|
caching off |
|
""" |
|
|
|
def __init__( |
|
self, |
|
use_listings_cache=True, |
|
listings_expiry_time=None, |
|
max_paths=None, |
|
**kwargs, |
|
): |
|
""" |
|
|
|
Parameters |
|
---------- |
|
use_listings_cache: bool |
|
If False, this cache never returns items, but always reports KeyError, |
|
and setting items has no effect |
|
listings_expiry_time: int or float (optional) |
|
Time in seconds that a listing is considered valid. If None, |
|
listings do not expire. |
|
max_paths: int (optional) |
|
The number of most recent listings that are considered valid; 'recent' |
|
refers to when the entry was set. |
|
""" |
|
self._cache = {} |
|
self._times = {} |
|
if max_paths: |
|
self._q = lru_cache(max_paths + 1)(lambda key: self._cache.pop(key, None)) |
|
self.use_listings_cache = use_listings_cache |
|
self.listings_expiry_time = listings_expiry_time |
|
self.max_paths = max_paths |
|
|
|
def __getitem__(self, item): |
|
if self.listings_expiry_time is not None: |
|
if self._times.get(item, 0) - time.time() < -self.listings_expiry_time: |
|
del self._cache[item] |
|
if self.max_paths: |
|
self._q(item) |
|
return self._cache[item] |
|
|
|
def clear(self): |
|
self._cache.clear() |
|
|
|
def __len__(self): |
|
return len(self._cache) |
|
|
|
def __contains__(self, item): |
|
try: |
|
self[item] |
|
return True |
|
except KeyError: |
|
return False |
|
|
|
def __setitem__(self, key, value): |
|
if not self.use_listings_cache: |
|
return |
|
if self.max_paths: |
|
self._q(key) |
|
self._cache[key] = value |
|
if self.listings_expiry_time is not None: |
|
self._times[key] = time.time() |
|
|
|
def __delitem__(self, key): |
|
del self._cache[key] |
|
|
|
def __iter__(self): |
|
entries = list(self._cache) |
|
|
|
return (k for k in entries if k in self) |
|
|
|
def __reduce__(self): |
|
return ( |
|
DirCache, |
|
(self.use_listings_cache, self.listings_expiry_time, self.max_paths), |
|
) |
|
|
