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