Coverage for locked_dict/locked_dict.py: 97.39%

1"""Provide a single class derived from dict to allow thread safe and mutable iterations through a lock.""" 

2 

3import itertools 

4import threading 

5from typing import no_type_check 

6 

7__all__ = ['LockedDict'] 

8 

9 

10# Use basestring name lookup test to adapt to python version 

11try: 

12 # noinspection PyCompatibility 

13 base_str = basestring # type: ignore 

14 items = 'iteritems' # from here on we are python < v3 

15except NameError: # ... below section applies to python v3+ 

16 base_str = str, bytes, bytearray 

17 items = 'items' 

18 

19 

20@no_type_check 

21class LockedDict(dict): # type: ignore 

22 """Work horse like dict derivative with an added lock to allow for 

23 context managed thread safe operation sections. 

24 Some methods added, so instances mostly behave like real dicts. 

25 Instances can be (de)serialized with pickle methods, given the 

26 lock state - skipped in dump and reload operations - is stable. 

27 Due to items attribute abstraction works with python v2.7+. 

28 """ 

29 

30 __slots__ = ('_lock',) # no __dict__ - that would be redundant 

31 

32 @no_type_check 

33 @staticmethod 

34 def _process_args(map_or_it=(), **kwargs): 

35 """Custom made helper for this class.""" 

36 if hasattr(map_or_it, items): 

37 map_or_it = getattr(map_or_it, items)() 

38 it_chain = itertools.chain 

39 return ((k, v) for k, v in it_chain(map_or_it, getattr(kwargs, items)())) 

40 

41 @no_type_check 

42 def __init__(self, mapping=(), **kwargs): 

43 """Base (dict) accepts mappings or iterables as first argument.""" 

44 super(LockedDict, self).__init__(self._process_args(mapping, **kwargs)) 

45 self._lock = threading.Lock() 

46 

47 @no_type_check 

48 def __enter__(self): 

49 """Context manager enter the block, acquire the lock.""" 

50 self._lock.acquire() 

51 return self 

52 

53 @no_type_check 

54 def __exit__(self, exc_type, exc_val, exc_tb): 

55 """Context manager exit the block, release the lock.""" 

56 self._lock.release() 

57 

58 @no_type_check 

59 def __getstate__(self): 

60 """Enable Pickling inside context blocks, 

61 through inclusion of the slot entries without the lock.""" 

62 return dict((slot, getattr(self, slot)) for slot in self.__slots__ if hasattr(self, slot) and slot != '_lock') 

63 

64 @no_type_check 

65 def __setstate__(self, state): 

66 """Restore the instance from pickle including the slot entries, 

67 without addition of a fresh lock. 

68 """ 

69 for slot, value in getattr(state, items)(): 69 ↛ 70line 69 didn't jump to line 70, because the loop on line 69 never started

70 setattr(self, slot, value) 

71 self._lock = threading.Lock() 

72 

73 @no_type_check 

74 def __getitem__(self, k): 

75 """For now plain delegation of getitem method to base class.""" 

76 return super(LockedDict, self).__getitem__(k) 

77 

78 @no_type_check 

79 def __setitem__(self, k, v): 

80 """For now plain delegation of setitem method to base class.""" 

81 return super(LockedDict, self).__setitem__(k, v) 

82 

83 @no_type_check 

84 def __delitem__(self, k): 

85 """For now plain delegation of del method to base class.""" 

86 return super(LockedDict, self).__delitem__(k) 

87 

88 @no_type_check 

89 def get(self, k, default=None): 

90 """For now plain delegation of get method to base class.""" 

91 return super(LockedDict, self).get(k, default) 

92 

93 @no_type_check 

94 def setdefault(self, k, default=None): 

95 """For now plain delegation of setdefault method to base class.""" 

96 return super(LockedDict, self).setdefault(k, default) 

97 

98 @no_type_check 

99 def pop(self, k, d=None): 

100 """For now plain delegation of pop method to base class.""" 

101 return super(LockedDict, self).pop(k, d) 

102 

103 @no_type_check 

104 def update(self, map_or_it=(), **kwargs): 

105 """Ensure processing of mappings or iterables as first argument.""" 

106 super(LockedDict, self).update(self._process_args(map_or_it, **kwargs)) 

107 

108 @no_type_check 

109 def __contains__(self, k): 

110 """For now plain delegation of contains method to base class.""" 

111 return super(LockedDict, self).__contains__(k) 

112 

113 @no_type_check 

114 @classmethod 

115 def fromkeys(cls, seq, value=None): 

116 """For now plain delegation of fromkeys class method to base.""" 

117 return super(LockedDict, cls).fromkeys(seq, value)