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

31 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2023-06-16 16:09 +0000

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

2# 

3# SPDX-License-Identifier: MIT 

4 

5import asyncio 

6 

7class TimeoutDict: 

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

9 item each refreshes the timeout. 

10 

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

12 

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

14 

15 This is not thread safe. 

16 """ 

17 

18 def __init__(self, timeout: float): 

19 self.timeout = timeout 

20 """Timeout set on any access 

21 

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

23 

24 self._items = {} 

25 """The actual dictionary""" 

26 self._recently_accessed = None 

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

28 self._timeout = None 

29 """Canceler for the timeout function""" 

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

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

32 

33 def __getitem__(self, key): 

34 result = self._items[key] 

35 self._accessed(key) 

36 return result 

37 

38 def __setitem__(self, key, value): 

39 self._items[key] = value 

40 self._accessed(key) 

41 

42 def _start_over(self): 

43 """Clear _recently_accessed, set the timeout""" 

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

45 self._recently_accessed = set() 

46 

47 def _accessed(self, key): 

48 """Mark a key as recently accessed""" 

49 if self._timeout is None: 

50 self._start_over() 

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

52 else: 

53 self._recently_accessed.add(key) 

54 

55 def _tick(self): 

56 self._items = {k: v for (k, v) in self._items.items() if k in self._recently_accessed} 

57 if self._items: 

58 self._start_over() 

59 else: 

60 self._timeout = None 

61 self._recently_accessed = None