Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/aio/_metadata.py: 37%

1# Copyright 2020 gRPC 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"""Implementation of the metadata abstraction for gRPC Asyncio Python.""" 

15from __future__ import annotations 

16 

17from collections import OrderedDict 

18from collections.abc import ( 

19 Collection, 

20 ItemsView, 

21 Iterable, 

22 Iterator, 

23 KeysView, 

24 Sequence, 

25 ValuesView, 

26) 

27from typing import Any, List, Optional, Tuple, Union 

28 

29from typing_extensions import Self 

30 

31MetadataKey = str 

32MetadataValue = Union[str, bytes] 

33MetadatumType = Tuple[MetadataKey, MetadataValue] 

34MetadataType = Union["Metadata", Sequence[MetadatumType]] 

35 

36 

37class Metadata(Collection): # noqa: PLW1641 

38 """Metadata abstraction for the asynchronous calls and interceptors. 

39 

40 The metadata is a mapping from str -> List[str] 

41 

42 Traits 

43 * Multiple entries are allowed for the same key 

44 * The order of the values by key is preserved 

45 * Getting by an element by key, retrieves the first mapped value 

46 * Supports an immutable view of the data 

47 * Allows partial mutation on the data without recreating the new object from scratch. 

48 """ 

49 

50 def __init__(self, *args: MetadatumType) -> None: 

51 self._metadata = OrderedDict() 

52 for md_key, md_value in args: 

53 self.add(md_key, md_value) 

54 

55 @classmethod 

56 def from_tuple(cls, raw_metadata: tuple): 

57 # Note: We unintentionally support non-tuple arguments here. We plan 

58 # to emit a DeprecationWarning when a non-tuple type is used. 

59 if raw_metadata: 

60 return cls(*raw_metadata) 

61 return cls() 

62 

63 @classmethod 

64 def _create( 

65 cls, 

66 raw_metadata: Union[None, Self, Iterable[MetadatumType]], 

67 ) -> Self: 

68 # TODO(asheshvidyut): Make this method public and encourage people to use it instead 

69 # of `from_tuple` to create metadata from non-tuple types. 

70 if raw_metadata is None: 

71 return cls() 

72 if isinstance(raw_metadata, cls): 

73 return raw_metadata 

74 if raw_metadata: 

75 return cls(*raw_metadata) 

76 return cls() 

77 

78 def add(self, key: MetadataKey, value: MetadataValue) -> None: 

79 self._metadata.setdefault(key, []) 

80 self._metadata[key].append(value) 

81 

82 def __len__(self) -> int: 

83 """Return the total number of elements that there are in the metadata, 

84 including multiple values for the same key. 

85 """ 

86 return sum(map(len, self._metadata.values())) 

87 

88 def __getitem__(self, key: MetadataKey) -> MetadataValue: 

89 """When calling <metadata>[<key>], the first element of all those 

90 mapped for <key> is returned. 

91 """ 

92 try: 

93 return self._metadata[key][0] 

94 except (ValueError, IndexError) as e: 

95 error_msg = f"{key!r}" 

96 raise KeyError(error_msg) from e 

97 

98 def __setitem__(self, key: MetadataKey, value: MetadataValue) -> None: 

99 """Calling metadata[<key>] = <value> 

100 Maps <value> to the first instance of <key>. 

101 """ 

102 if key not in self: 

103 self._metadata[key] = [value] 

104 else: 

105 current_values = self.get_all(key) 

106 self._metadata[key] = [value, *current_values[1:]] 

107 

108 def __delitem__(self, key: MetadataKey) -> None: 

109 """``del metadata[<key>]`` deletes the first mapping for <key>.""" 

110 current_values = self.get_all(key) 

111 if not current_values: 

112 raise KeyError(repr(key)) 

113 self._metadata[key] = current_values[1:] 

114 

115 def delete_all(self, key: MetadataKey) -> None: 

116 """Delete all mappings for <key>.""" 

117 del self._metadata[key] 

118 

119 def __iter__(self) -> Iterator[Tuple[MetadataKey, MetadataValue]]: 

120 for key, values in self._metadata.items(): 

121 for value in values: 

122 yield (key, value) 

123 

124 def keys(self) -> KeysView: 

125 return KeysView(self._metadata) 

126 

127 def values(self) -> ValuesView: 

128 return ValuesView(self._metadata) 

129 

130 def items(self) -> ItemsView: 

131 return ItemsView(self._metadata) 

132 

133 def get( 

134 self, key: MetadataKey, default: Optional[MetadataValue] = None 

135 ) -> Optional[MetadataValue]: 

136 try: 

137 return self[key] 

138 except KeyError: 

139 return default 

140 

141 def get_all(self, key: MetadataKey) -> List[MetadataValue]: 

142 """For compatibility with other Metadata abstraction objects (like in Java), 

143 this would return all items under the desired <key>. 

144 """ 

145 return self._metadata.get(key, []) 

146 

147 def set_all(self, key: MetadataKey, values: List[MetadataValue]) -> None: 

148 self._metadata[key] = values 

149 

150 def __contains__(self, key: MetadataKey) -> bool: 

151 return key in self._metadata 

152 

153 def __eq__(self, other: object) -> bool: 

154 if isinstance(other, self.__class__): 

155 return self._metadata == other._metadata 

156 if isinstance(other, tuple): 

157 return tuple(self) == other 

158 return NotImplemented # pytype: disable=bad-return-type 

159 

160 def __add__(self, other: Any) -> "Metadata": 

161 if isinstance(other, self.__class__): 

162 return Metadata(*(tuple(self) + tuple(other))) 

163 if isinstance(other, tuple): 

164 return Metadata(*(tuple(self) + other)) 

165 return NotImplemented # pytype: disable=bad-return-type 

166 

167 def __repr__(self) -> str: 

168 view = tuple(self) 

169 return "{0}({1!r})".format(self.__class__.__name__, view)