Coverage for aiocoap/util/asyncio/timeoutdict.py: 100%

1# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors 

2# 

3# SPDX-License-Identifier: MIT 

4 

5import asyncio 

6from typing import Dict, Generic, TypeVar 

7 

8K = TypeVar("K") 

9V = TypeVar("V") 

10 

11 

12class TimeoutDict(Generic[K, V]): 

13 """A dict-ish type whose entries live on a timeout; adding and accessing an 

14 item each refreshes the timeout. 

15 

16 The timeout is a lower bound; items may live up to twice as long. 

17 

18 The container is implemented incompletely, with additions made on demand. 

19 

20 This is not thread safe. 

21 """ 

22 

23 def __init__(self, timeout: float): 

24 self.timeout = timeout 

25 """Timeout set on any access 

26 

27 This can be changed at runtime, but changes only take effect """ 

28 

29 self._items: Dict[K, V] = {} 

30 """The actual dictionary""" 

31 self._recently_accessed = None 

32 """Items accessed since the timeout last fired""" 

33 self._timeout = None 

34 """Canceler for the timeout function""" 

35 # Note: Without a __del__ implementation that even cancels, the object 

36 # will be kept alive by the main loop for a timeout 

37 

38 def __getitem__(self, key): 

39 result = self._items[key] 

40 self._accessed(key) 

41 return result 

42 

43 def __setitem__(self, key, value): 

44 self._items[key] = value 

45 self._accessed(key) 

46 

47 def _start_over(self): 

48 """Clear _recently_accessed, set the timeout""" 

49 self._timeout = asyncio.get_running_loop().call_later(self.timeout, self._tick) 

50 self._recently_accessed = set() 

51 

52 def _accessed(self, key): 

53 """Mark a key as recently accessed""" 

54 if self._timeout is None: 

55 self._start_over() 

56 # No need to add the key, it'll live for this duration anyway 

57 else: 

58 self._recently_accessed.add(key) 

59 

60 def _tick(self): 

61 self._items = { 

62 k: v for (k, v) in self._items.items() if k in self._recently_accessed 

63 } 

64 if self._items: 

65 self._start_over() 

66 else: 

67 self._timeout = None 

68 self._recently_accessed = None