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

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 Collection 

19from collections.abc import ItemsView 

20from collections.abc import Iterable 

21from collections.abc import Iterator 

22from collections.abc import KeysView 

23from collections.abc import Sequence 

24from collections.abc import ValuesView 

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

26 

27from typing_extensions import Self 

28 

29MetadataKey = str 

30MetadataValue = Union[str, bytes] 

31MetadatumType = Tuple[MetadataKey, MetadataValue] 

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

33 

34 

35class Metadata(Collection): # noqa: PLW1641 

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

37 

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

39 

40 Traits 

41 * Multiple entries are allowed for the same key 

42 * The order of the values by key is preserved 

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

44 * Supports an immutable view of the data 

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

46 """ 

47 

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

49 self._metadata = OrderedDict() 

50 for md_key, md_value in args: 

51 self.add(md_key, md_value) 

52 

53 @classmethod 

54 def from_tuple(cls, raw_metadata: tuple): 

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

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

57 if raw_metadata: 

58 return cls(*raw_metadata) 

59 return cls() 

60 

61 @classmethod 

62 def _create( 

63 cls, 

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

65 ) -> Self: 

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

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

68 if raw_metadata is None: 

69 return Metadata() 

70 if isinstance(raw_metadata, cls): 

71 return raw_metadata 

72 if raw_metadata: 

73 return cls(*raw_metadata) 

74 return cls() 

75 

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

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

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

79 

80 def __len__(self) -> int: 

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

82 including multiple values for the same key. 

83 """ 

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

85 

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

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

88 mapped for <key> is returned. 

89 """ 

90 try: 

91 return self._metadata[key][0] 

92 except (ValueError, IndexError) as e: 

93 error_msg = f"{key!r}" 

94 raise KeyError(error_msg) from e 

95 

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

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

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

99 """ 

100 if key not in self: 

101 self._metadata[key] = [value] 

102 else: 

103 current_values = self.get_all(key) 

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

105 

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

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

108 current_values = self.get_all(key) 

109 if not current_values: 

110 raise KeyError(repr(key)) 

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

112 

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

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

115 del self._metadata[key] 

116 

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

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

119 for value in values: 

120 yield (key, value) 

121 

122 def keys(self) -> KeysView: 

123 return KeysView(self._metadata) 

124 

125 def values(self) -> ValuesView: 

126 return ValuesView(self._metadata) 

127 

128 def items(self) -> ItemsView: 

129 return ItemsView(self._metadata) 

130 

131 def get( 

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

133 ) -> Optional[MetadataValue]: 

134 try: 

135 return self[key] 

136 except KeyError: 

137 return default 

138 

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

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

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

142 """ 

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

144 

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

146 self._metadata[key] = values 

147 

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

149 return key in self._metadata 

150 

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

152 if isinstance(other, self.__class__): 

153 return self._metadata == other._metadata 

154 if isinstance(other, tuple): 

155 return tuple(self) == other 

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

157 

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

159 if isinstance(other, self.__class__): 

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

161 if isinstance(other, tuple): 

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

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

164 

165 def __repr__(self) -> str: 

166 view = tuple(self) 

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