Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/fsspec/dircache.py: 43%

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 entries = list(self._cache) 

91 

92 return (k for k in entries if k in self) 

93 

94 def __reduce__(self): 

95 return ( 

96 DirCache, 

97 (self.use_listings_cache, self.listings_expiry_time, self.max_paths), 

98 )