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

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 collections import OrderedDict 

16from collections import abc 

17from typing import Any, Iterator, List, Optional, Tuple, Union 

18 

19MetadataKey = str 

20MetadataValue = Union[str, bytes] 

21 

22 

23class Metadata(abc.Collection): 

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

25 

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

27 

28 Traits 

29 * Multiple entries are allowed for the same key 

30 * The order of the values by key is preserved 

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

32 * Supports an immutable view of the data 

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

34 """ 

35 

36 def __init__(self, *args: Tuple[MetadataKey, MetadataValue]) -> None: 

37 self._metadata = OrderedDict() 

38 for md_key, md_value in args: 

39 self.add(md_key, md_value) 

40 

41 @classmethod 

42 def from_tuple(cls, raw_metadata: tuple): 

43 if raw_metadata: 

44 return cls(*raw_metadata) 

45 return cls() 

46 

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

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

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

50 

51 def __len__(self) -> int: 

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

53 including multiple values for the same key. 

54 """ 

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

56 

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

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

59 mapped for <key> is returned. 

60 """ 

61 try: 

62 return self._metadata[key][0] 

63 except (ValueError, IndexError) as e: 

64 error_msg = f"{key!r}" 

65 raise KeyError(error_msg) from e 

66 

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

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

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

70 """ 

71 if key not in self: 

72 self._metadata[key] = [value] 

73 else: 

74 current_values = self.get_all(key) 

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

76 

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

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

79 current_values = self.get_all(key) 

80 if not current_values: 

81 raise KeyError(repr(key)) 

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

83 

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

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

86 del self._metadata[key] 

87 

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

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

90 for value in values: 

91 yield (key, value) 

92 

93 def keys(self) -> abc.KeysView: 

94 return abc.KeysView(self) 

95 

96 def values(self) -> abc.ValuesView: 

97 return abc.ValuesView(self) 

98 

99 def items(self) -> abc.ItemsView: 

100 return abc.ItemsView(self) 

101 

102 def get( 

103 self, key: MetadataKey, default: MetadataValue = None 

104 ) -> Optional[MetadataValue]: 

105 try: 

106 return self[key] 

107 except KeyError: 

108 return default 

109 

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

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

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

113 """ 

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

115 

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

117 self._metadata[key] = values 

118 

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

120 return key in self._metadata 

121 

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

123 if isinstance(other, self.__class__): 

124 return self._metadata == other._metadata 

125 if isinstance(other, tuple): 

126 return tuple(self) == other 

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

128 

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

130 if isinstance(other, self.__class__): 

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

132 if isinstance(other, tuple): 

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

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

135 

136 def __repr__(self) -> str: 

137 view = tuple(self) 

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