Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/opentelemetry/sdk/util/__init__.py: 35%

1# Copyright The OpenTelemetry Authors 

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 

15import datetime 

16import threading 

17from collections import OrderedDict, deque 

18from collections.abc import MutableMapping, Sequence 

19from typing import Optional 

20 

21from deprecated import deprecated 

22 

23 

24def ns_to_iso_str(nanoseconds): 

25 """Get an ISO 8601 string from time_ns value.""" 

26 ts = datetime.datetime.utcfromtimestamp(nanoseconds / 1e9) 

27 return ts.strftime("%Y-%m-%dT%H:%M:%S.%fZ") 

28 

29 

30def get_dict_as_key(labels): 

31 """Converts a dict to be used as a unique key""" 

32 return tuple( 

33 sorted( 

34 map( 

35 lambda kv: (kv[0], tuple(kv[1])) 

36 if isinstance(kv[1], list) 

37 else kv, 

38 labels.items(), 

39 ) 

40 ) 

41 ) 

42 

43 

44class BoundedList(Sequence): 

45 """An append only list with a fixed max size. 

46 

47 Calls to `append` and `extend` will drop the oldest elements if there is 

48 not enough room. 

49 """ 

50 

51 def __init__(self, maxlen: Optional[int]): 

52 self.dropped = 0 

53 self._dq = deque(maxlen=maxlen) # type: deque 

54 self._lock = threading.Lock() 

55 

56 def __repr__(self): 

57 return f"{type(self).__name__}({list(self._dq)}, maxlen={self._dq.maxlen})" 

58 

59 def __getitem__(self, index): 

60 return self._dq[index] 

61 

62 def __len__(self): 

63 return len(self._dq) 

64 

65 def __iter__(self): 

66 with self._lock: 

67 return iter(deque(self._dq)) 

68 

69 def append(self, item): 

70 with self._lock: 

71 if ( 

72 self._dq.maxlen is not None 

73 and len(self._dq) == self._dq.maxlen 

74 ): 

75 self.dropped += 1 

76 self._dq.append(item) 

77 

78 def extend(self, seq): 

79 with self._lock: 

80 if self._dq.maxlen is not None: 

81 to_drop = len(seq) + len(self._dq) - self._dq.maxlen 

82 if to_drop > 0: 

83 self.dropped += to_drop 

84 self._dq.extend(seq) 

85 

86 @classmethod 

87 def from_seq(cls, maxlen, seq): 

88 seq = tuple(seq) 

89 bounded_list = cls(maxlen) 

90 bounded_list.extend(seq) 

91 return bounded_list 

92 

93 

94@deprecated(version="1.4.0") # type: ignore 

95class BoundedDict(MutableMapping): 

96 """An ordered dict with a fixed max capacity. 

97 

98 Oldest elements are dropped when the dict is full and a new element is 

99 added. 

100 """ 

101 

102 def __init__(self, maxlen: Optional[int]): 

103 if maxlen is not None: 

104 if not isinstance(maxlen, int): 

105 raise ValueError 

106 if maxlen < 0: 

107 raise ValueError 

108 self.maxlen = maxlen 

109 self.dropped = 0 

110 self._dict = OrderedDict() # type: OrderedDict 

111 self._lock = threading.Lock() # type: threading.Lock 

112 

113 def __repr__(self): 

114 return ( 

115 f"{type(self).__name__}({dict(self._dict)}, maxlen={self.maxlen})" 

116 ) 

117 

118 def __getitem__(self, key): 

119 return self._dict[key] 

120 

121 def __setitem__(self, key, value): 

122 with self._lock: 

123 if self.maxlen is not None and self.maxlen == 0: 

124 self.dropped += 1 

125 return 

126 

127 if key in self._dict: 

128 del self._dict[key] 

129 elif self.maxlen is not None and len(self._dict) == self.maxlen: 

130 del self._dict[next(iter(self._dict.keys()))] 

131 self.dropped += 1 

132 self._dict[key] = value 

133 

134 def __delitem__(self, key): 

135 del self._dict[key] 

136 

137 def __iter__(self): 

138 with self._lock: 

139 return iter(self._dict.copy()) 

140 

141 def __len__(self): 

142 return len(self._dict) 

143 

144 @classmethod 

145 def from_map(cls, maxlen, mapping): 

146 mapping = OrderedDict(mapping) 

147 bounded_dict = cls(maxlen) 

148 for key, value in mapping.items(): 

149 bounded_dict[key] = value 

150 return bounded_dict