Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/_common.py: 45%

1# Copyright 2016 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"""Shared implementation.""" 

15 

16import logging 

17import time 

18from typing import Any, AnyStr, Callable, Optional, Union 

19 

20import grpc 

21from grpc._cython import cygrpc 

22from grpc._typing import DeserializingFunction 

23from grpc._typing import SerializingFunction 

24 

25_LOGGER = logging.getLogger(__name__) 

26 

27CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { 

28 cygrpc.ConnectivityState.idle: grpc.ChannelConnectivity.IDLE, 

29 cygrpc.ConnectivityState.connecting: grpc.ChannelConnectivity.CONNECTING, 

30 cygrpc.ConnectivityState.ready: grpc.ChannelConnectivity.READY, 

31 cygrpc.ConnectivityState.transient_failure: grpc.ChannelConnectivity.TRANSIENT_FAILURE, 

32 cygrpc.ConnectivityState.shutdown: grpc.ChannelConnectivity.SHUTDOWN, 

33} 

34 

35CYGRPC_STATUS_CODE_TO_STATUS_CODE = { 

36 cygrpc.StatusCode.ok: grpc.StatusCode.OK, 

37 cygrpc.StatusCode.cancelled: grpc.StatusCode.CANCELLED, 

38 cygrpc.StatusCode.unknown: grpc.StatusCode.UNKNOWN, 

39 cygrpc.StatusCode.invalid_argument: grpc.StatusCode.INVALID_ARGUMENT, 

40 cygrpc.StatusCode.deadline_exceeded: grpc.StatusCode.DEADLINE_EXCEEDED, 

41 cygrpc.StatusCode.not_found: grpc.StatusCode.NOT_FOUND, 

42 cygrpc.StatusCode.already_exists: grpc.StatusCode.ALREADY_EXISTS, 

43 cygrpc.StatusCode.permission_denied: grpc.StatusCode.PERMISSION_DENIED, 

44 cygrpc.StatusCode.unauthenticated: grpc.StatusCode.UNAUTHENTICATED, 

45 cygrpc.StatusCode.resource_exhausted: grpc.StatusCode.RESOURCE_EXHAUSTED, 

46 cygrpc.StatusCode.failed_precondition: grpc.StatusCode.FAILED_PRECONDITION, 

47 cygrpc.StatusCode.aborted: grpc.StatusCode.ABORTED, 

48 cygrpc.StatusCode.out_of_range: grpc.StatusCode.OUT_OF_RANGE, 

49 cygrpc.StatusCode.unimplemented: grpc.StatusCode.UNIMPLEMENTED, 

50 cygrpc.StatusCode.internal: grpc.StatusCode.INTERNAL, 

51 cygrpc.StatusCode.unavailable: grpc.StatusCode.UNAVAILABLE, 

52 cygrpc.StatusCode.data_loss: grpc.StatusCode.DATA_LOSS, 

53} 

54STATUS_CODE_TO_CYGRPC_STATUS_CODE = { 

55 grpc_code: cygrpc_code 

56 for cygrpc_code, grpc_code in CYGRPC_STATUS_CODE_TO_STATUS_CODE.items() 

57} 

58 

59MAXIMUM_WAIT_TIMEOUT = 0.1 

60 

61_ERROR_MESSAGE_PORT_BINDING_FAILED = ( 

62 "Failed to bind to address %s; set " 

63 "GRPC_VERBOSITY=debug environment variable to see detailed error message." 

64) 

65 

66 

67def encode(s: AnyStr) -> bytes: 

68 if isinstance(s, bytes): 

69 return s 

70 return s.encode("utf8") 

71 

72 

73def decode(b: AnyStr) -> str: 

74 if isinstance(b, bytes): 

75 return b.decode("utf-8", "replace") 

76 return b 

77 

78 

79def _transform( 

80 message: Any, 

81 transformer: Union[SerializingFunction, DeserializingFunction, None], 

82 exception_message: str, 

83) -> Any: 

84 if transformer is None: 

85 return message 

86 try: 

87 return transformer(message) 

88 except Exception: # pylint: disable=broad-except 

89 _LOGGER.exception(exception_message) 

90 return None 

91 

92 

93def serialize(message: Any, serializer: Optional[SerializingFunction]) -> bytes: 

94 return _transform(message, serializer, "Exception serializing message!") 

95 

96 

97def deserialize( 

98 serialized_message: bytes, deserializer: Optional[DeserializingFunction] 

99) -> Any: 

100 return _transform( 

101 serialized_message, deserializer, "Exception deserializing message!" 

102 ) 

103 

104 

105def fully_qualified_method(group: str, method: str) -> str: 

106 return "/{}/{}".format(group, method) 

107 

108 

109def _wait_once( 

110 wait_fn: Callable[..., bool], 

111 timeout: float, 

112 spin_cb: Optional[Callable[[], None]], 

113): 

114 wait_fn(timeout=timeout) 

115 if spin_cb is not None: 

116 spin_cb() 

117 

118 

119def wait( 

120 wait_fn: Callable[..., bool], 

121 wait_complete_fn: Callable[[], bool], 

122 timeout: Optional[float] = None, 

123 spin_cb: Optional[Callable[[], None]] = None, 

124) -> bool: 

125 """Blocks waiting for an event without blocking the thread indefinitely. 

126 

127 See https://github.com/grpc/grpc/issues/19464 for full context. CPython's 

128 `threading.Event.wait` and `threading.Condition.wait` methods, if invoked 

129 without a timeout kwarg, may block the calling thread indefinitely. If the 

130 call is made from the main thread, this means that signal handlers may not 

131 run for an arbitrarily long period of time. 

132 

133 This wrapper calls the supplied wait function with an arbitrary short 

134 timeout to ensure that no signal handler has to wait longer than 

135 MAXIMUM_WAIT_TIMEOUT before executing. 

136 

137 Args: 

138 wait_fn: A callable acceptable a single float-valued kwarg named 

139 `timeout`. This function is expected to be one of `threading.Event.wait` 

140 or `threading.Condition.wait`. 

141 wait_complete_fn: A callable taking no arguments and returning a bool. 

142 When this function returns true, it indicates that waiting should cease. 

143 timeout: An optional float-valued number of seconds after which the wait 

144 should cease. 

145 spin_cb: An optional Callable taking no arguments and returning nothing. 

146 This callback will be called on each iteration of the spin. This may be 

147 used for, e.g. work related to forking. 

148 

149 Returns: 

150 True if a timeout was supplied and it was reached. False otherwise. 

151 """ 

152 if timeout is None: 

153 while not wait_complete_fn(): 

154 _wait_once(wait_fn, MAXIMUM_WAIT_TIMEOUT, spin_cb) 

155 else: 

156 end = time.time() + timeout 

157 while not wait_complete_fn(): 

158 remaining = min(end - time.time(), MAXIMUM_WAIT_TIMEOUT) 

159 if remaining < 0: 

160 return True 

161 _wait_once(wait_fn, remaining, spin_cb) 

162 return False 

163 

164 

165def validate_port_binding_result(address: str, port: int) -> int: 

166 """Validates if the port binding succeed. 

167 

168 If the port returned by Core is 0, the binding is failed. However, in that 

169 case, the Core API doesn't return a detailed failing reason. The best we 

170 can do is raising an exception to prevent further confusion. 

171 

172 Args: 

173 address: The address string to be bound. 

174 port: An int returned by core 

175 """ 

176 if port == 0: 

177 # The Core API doesn't return a failure message. The best we can do 

178 # is raising an exception to prevent further confusion. 

179 raise RuntimeError(_ERROR_MESSAGE_PORT_BINDING_FAILED % address) 

180 else: 

181 return port