1import time 2from collections.abc import MutableMapping 3from functools import lru_cache 4 5 6class DirCache(MutableMapping): 7 """ 8 Caching of directory listings, in a structure like:: 9 10 {"path0": [ 11 {"name": "path0/file0", 12 "size": 123, 13 "type": "file", 14 ... 15 }, 16 {"name": "path0/file1", 17 }, 18 ... 19 ], 20 "path1": [...] 21 } 22 23 Parameters to this class control listing expiry or indeed turn 24 caching off 25 """ 26 27 def __init__( 28 self, 29 use_listings_cache=True, 30 listings_expiry_time=None, 31 max_paths=None, 32 **kwargs, 33 ): 34 """ 35 36 Parameters 37 ---------- 38 use_listings_cache: bool 39 If False, this cache never returns items, but always reports KeyError, 40 and setting items has no effect 41 listings_expiry_time: int or float (optional) 42 Time in seconds that a listing is considered valid. If None, 43 listings do not expire. 44 max_paths: int (optional) 45 The number of most recent listings that are considered valid; 'recent' 46 refers to when the entry was set. 47 """ 48 self._cache = {} 49 self._times = {} 50 if max_paths: 51 self._q = lru_cache(max_paths + 1)(lambda key: self._cache.pop(key, None)) 52 self.use_listings_cache = use_listings_cache 53 self.listings_expiry_time = listings_expiry_time 54 self.max_paths = max_paths 55 56 def __getitem__(self, item): 57 if self.listings_expiry_time is not None: 58 if self._times.get(item, 0) - time.time() < -self.listings_expiry_time: 59 del self._cache[item] 60 if self.max_paths: 61 self._q(item) 62 return self._cache[item] # maybe raises KeyError 63 64 def clear(self): 65 self._cache.clear() 66 67 def __len__(self): 68 return len(self._cache) 69 70 def __contains__(self, item): 71 try: 72 self[item] 73 return True 74 except KeyError: 75 return False 76 77 def __setitem__(self, key, value): 78 if not self.use_listings_cache: 79 return 80 if self.max_paths: 81 self._q(key) 82 self._cache[key] = value 83 if self.listings_expiry_time is not None: 84 self._times[key] = time.time() 85 86 def __delitem__(self, key): 87 del self._cache[key] 88 89 def __iter__(self): 90 return (k for k in self._cache if k in self) 91 92 def __reduce__(self): 93 return ( 94 DirCache, 95 (self.use_listings_cache, self.listings_expiry_time, self.max_paths), 96 ) 97