Coverage for /builds/BuildGrid/buildgrid/buildgrid/server/utils/lru_inmemory_cache.py: 96.15%

26 statements  

« prev     ^ index     » next       coverage.py v6.4.3, created at 2022-08-09 22:05 +0000

1# Copyright (C) 2021 Bloomberg LP 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# <http://www.apache.org/licenses/LICENSE-2.0> 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15from collections import OrderedDict 

16 

17from threading import Lock 

18 

19 

20class LruInMemoryCache: 

21 def __init__(self, capacity: int): 

22 """Simple in-memory LRU storage container. 

23 

24 Args: 

25 capacity (int): the maximum number of entries that can be cached. 

26 After exceeding the capacity, older entries will be dropped 

27 to make room for new ones, following a Least Recently Used 

28 policy. 

29 """ 

30 if capacity <= 0: 

31 raise ValueError("Capacity must be positive") 

32 

33 self._capacity = capacity 

34 

35 self._ordered_dict_lock = Lock() 

36 self._ordered_dict = OrderedDict() # type: ignore 

37 

38 @property 

39 def size(self): 

40 """Number of elements in the cache.""" 

41 with self._ordered_dict_lock: 

42 return len(self._ordered_dict) 

43 

44 @property 

45 def capacity(self): 

46 """Maximum number of elements that fit in the cache.""" 

47 return self._capacity 

48 

49 def update(self, key, value): 

50 """Update a cache entry.""" 

51 with self._ordered_dict_lock: 

52 self._ordered_dict[key] = value 

53 self._ordered_dict.move_to_end(key, last=True) 

54 

55 if len(self._ordered_dict) > self._capacity: 

56 self._ordered_dict.popitem(last=False) 

57 

58 def get(self, key): 

59 """Get a value defined in the cache. If a value for the given key is 

60 not found, returns None. 

61 Updates the last access time of the entry. 

62 """ 

63 with self._ordered_dict_lock: 

64 value = self._ordered_dict.get(key, None) 

65 if value is not None: 

66 self._ordered_dict.move_to_end(key, last=True) 

67 return value