Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/websockets/datastructures.py: 58%

59 statements  

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

1from __future__ import annotations 

2 

3from typing import ( 

4 Any, 

5 Dict, 

6 Iterable, 

7 Iterator, 

8 List, 

9 Mapping, 

10 MutableMapping, 

11 Protocol, 

12 Tuple, 

13 Union, 

14) 

15 

16 

17__all__ = ["Headers", "HeadersLike", "MultipleValuesError"] 

18 

19 

20class MultipleValuesError(LookupError): 

21 """ 

22 Exception raised when :class:`Headers` has more than one value for a key. 

23 

24 """ 

25 

26 def __str__(self) -> str: 

27 # Implement the same logic as KeyError_str in Objects/exceptions.c. 

28 if len(self.args) == 1: 

29 return repr(self.args[0]) 

30 return super().__str__() 

31 

32 

33class Headers(MutableMapping[str, str]): 

34 """ 

35 Efficient data structure for manipulating HTTP headers. 

36 

37 A :class:`list` of ``(name, values)`` is inefficient for lookups. 

38 

39 A :class:`dict` doesn't suffice because header names are case-insensitive 

40 and multiple occurrences of headers with the same name are possible. 

41 

42 :class:`Headers` stores HTTP headers in a hybrid data structure to provide 

43 efficient insertions and lookups while preserving the original data. 

44 

45 In order to account for multiple values with minimal hassle, 

46 :class:`Headers` follows this logic: 

47 

48 - When getting a header with ``headers[name]``: 

49 - if there's no value, :exc:`KeyError` is raised; 

50 - if there's exactly one value, it's returned; 

51 - if there's more than one value, :exc:`MultipleValuesError` is raised. 

52 

53 - When setting a header with ``headers[name] = value``, the value is 

54 appended to the list of values for that header. 

55 

56 - When deleting a header with ``del headers[name]``, all values for that 

57 header are removed (this is slow). 

58 

59 Other methods for manipulating headers are consistent with this logic. 

60 

61 As long as no header occurs multiple times, :class:`Headers` behaves like 

62 :class:`dict`, except keys are lower-cased to provide case-insensitivity. 

63 

64 Two methods support manipulating multiple values explicitly: 

65 

66 - :meth:`get_all` returns a list of all values for a header; 

67 - :meth:`raw_items` returns an iterator of ``(name, values)`` pairs. 

68 

69 """ 

70 

71 __slots__ = ["_dict", "_list"] 

72 

73 # Like dict, Headers accepts an optional "mapping or iterable" argument. 

74 def __init__(self, *args: HeadersLike, **kwargs: str) -> None: 

75 self._dict: Dict[str, List[str]] = {} 

76 self._list: List[Tuple[str, str]] = [] 

77 self.update(*args, **kwargs) 

78 

79 def __str__(self) -> str: 

80 return "".join(f"{key}: {value}\r\n" for key, value in self._list) + "\r\n" 

81 

82 def __repr__(self) -> str: 

83 return f"{self.__class__.__name__}({self._list!r})" 

84 

85 def copy(self) -> Headers: 

86 copy = self.__class__() 

87 copy._dict = self._dict.copy() 

88 copy._list = self._list.copy() 

89 return copy 

90 

91 def serialize(self) -> bytes: 

92 # Since headers only contain ASCII characters, we can keep this simple. 

93 return str(self).encode() 

94 

95 # Collection methods 

96 

97 def __contains__(self, key: object) -> bool: 

98 return isinstance(key, str) and key.lower() in self._dict 

99 

100 def __iter__(self) -> Iterator[str]: 

101 return iter(self._dict) 

102 

103 def __len__(self) -> int: 

104 return len(self._dict) 

105 

106 # MutableMapping methods 

107 

108 def __getitem__(self, key: str) -> str: 

109 value = self._dict[key.lower()] 

110 if len(value) == 1: 

111 return value[0] 

112 else: 

113 raise MultipleValuesError(key) 

114 

115 def __setitem__(self, key: str, value: str) -> None: 

116 self._dict.setdefault(key.lower(), []).append(value) 

117 self._list.append((key, value)) 

118 

119 def __delitem__(self, key: str) -> None: 

120 key_lower = key.lower() 

121 self._dict.__delitem__(key_lower) 

122 # This is inefficient. Fortunately deleting HTTP headers is uncommon. 

123 self._list = [(k, v) for k, v in self._list if k.lower() != key_lower] 

124 

125 def __eq__(self, other: Any) -> bool: 

126 if not isinstance(other, Headers): 

127 return NotImplemented 

128 return self._dict == other._dict 

129 

130 def clear(self) -> None: 

131 """ 

132 Remove all headers. 

133 

134 """ 

135 self._dict = {} 

136 self._list = [] 

137 

138 def update(self, *args: HeadersLike, **kwargs: str) -> None: 

139 """ 

140 Update from a :class:`Headers` instance and/or keyword arguments. 

141 

142 """ 

143 args = tuple( 

144 arg.raw_items() if isinstance(arg, Headers) else arg for arg in args 

145 ) 

146 super().update(*args, **kwargs) 

147 

148 # Methods for handling multiple values 

149 

150 def get_all(self, key: str) -> List[str]: 

151 """ 

152 Return the (possibly empty) list of all values for a header. 

153 

154 Args: 

155 key: header name. 

156 

157 """ 

158 return self._dict.get(key.lower(), []) 

159 

160 def raw_items(self) -> Iterator[Tuple[str, str]]: 

161 """ 

162 Return an iterator of all values as ``(name, value)`` pairs. 

163 

164 """ 

165 return iter(self._list) 

166 

167 

168# copy of _typeshed.SupportsKeysAndGetItem. 

169class SupportsKeysAndGetItem(Protocol): # pragma: no cover 

170 """ 

171 Dict-like types with ``keys() -> str`` and ``__getitem__(key: str) -> str`` methods. 

172 

173 """ 

174 

175 def keys(self) -> Iterable[str]: 

176 ... 

177 

178 def __getitem__(self, key: str) -> str: 

179 ... 

180 

181 

182HeadersLike = Union[ 

183 Headers, 

184 Mapping[str, str], 

185 Iterable[Tuple[str, str]], 

186 SupportsKeysAndGetItem, 

187] 

188""" 

189Types accepted where :class:`Headers` is expected. 

190 

191In addition to :class:`Headers` itself, this includes dict-like types where both 

192keys and values are :class:`str`. 

193 

194"""