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

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 else: 

71 return s.encode("utf8") 

72 

73 

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

75 if isinstance(b, bytes): 

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

77 return b 

78 

79 

80def _transform( 

81 message: Any, 

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

83 exception_message: str, 

84) -> Any: 

85 if transformer is None: 

86 return message 

87 else: 

88 try: 

89 return transformer(message) 

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

91 _LOGGER.exception(exception_message) 

92 return None 

93 

94 

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

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

97 

98 

99def deserialize( 

100 serialized_message: bytes, deserializer: Optional[DeserializingFunction] 

101) -> Any: 

102 return _transform( 

103 serialized_message, deserializer, "Exception deserializing message!" 

104 ) 

105 

106 

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

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

109 

110 

111def _wait_once( 

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

113 timeout: float, 

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

115): 

116 wait_fn(timeout=timeout) 

117 if spin_cb is not None: 

118 spin_cb() 

119 

120 

121def wait( 

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

123 wait_complete_fn: Callable[[], bool], 

124 timeout: Optional[float] = None, 

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

126) -> bool: 

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

128 

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

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

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

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

133 run for an arbitrarily long period of time. 

134 

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

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

137 MAXIMUM_WAIT_TIMEOUT before executing. 

138 

139 Args: 

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

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

142 or `threading.Condition.wait`. 

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

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

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

146 should cease. 

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

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

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

150 

151 Returns: 

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

153 """ 

154 if timeout is None: 

155 while not wait_complete_fn(): 

156 _wait_once(wait_fn, MAXIMUM_WAIT_TIMEOUT, spin_cb) 

157 else: 

158 end = time.time() + timeout 

159 while not wait_complete_fn(): 

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

161 if remaining < 0: 

162 return True 

163 _wait_once(wait_fn, remaining, spin_cb) 

164 return False 

165 

166 

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

168 """Validates if the port binding succeed. 

169 

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

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

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

173 

174 Args: 

175 address: The address string to be bound. 

176 port: An int returned by core 

177 """ 

178 if port == 0: 

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

180 # is raising an exception to prevent further confusion. 

181 raise RuntimeError(_ERROR_MESSAGE_PORT_BINDING_FAILED % address) 

182 else: 

183 return port