Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc_status/rpc_status.py: 54%

1# Copyright 2018 The 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"""Reference implementation for status mapping in gRPC Python.""" 

15 

16import collections 

17import sys 

18 

19from google.rpc import status_pb2 

20import grpc 

21 

22from ._common import GRPC_DETAILS_METADATA_KEY 

23from ._common import code_to_grpc_status_code 

24 

25 

26class _Status( 

27 collections.namedtuple("_Status", ("code", "details", "trailing_metadata")), 

28 grpc.Status, 

29): 

30 pass 

31 

32 

33def from_call(call): 

34 """Returns a google.rpc.status.Status message corresponding to a given grpc.Call. 

35 

36 This is an EXPERIMENTAL API. 

37 

38 Args: 

39 call: A grpc.Call instance. 

40 

41 Returns: 

42 A google.rpc.status.Status message representing the status of the RPC. 

43 

44 Raises: 

45 ValueError: If the gRPC call's code or details are inconsistent with the 

46 status code and message inside of the google.rpc.status.Status. 

47 """ 

48 if call.trailing_metadata() is None: 

49 return None 

50 for key, value in call.trailing_metadata(): 

51 if key == GRPC_DETAILS_METADATA_KEY: 

52 rich_status = status_pb2.Status.FromString(value) 

53 if call.code().value[0] != rich_status.code: 

54 raise ValueError( 

55 "Code in Status proto (%s) doesn't match status code (%s)" 

56 % (code_to_grpc_status_code(rich_status.code), call.code()) 

57 ) 

58 if call.details() != rich_status.message: 

59 raise ValueError( 

60 "Message in Status proto (%s) doesn't match status details" 

61 " (%s)" % (rich_status.message, call.details()) 

62 ) 

63 return rich_status 

64 return None 

65 

66 

67def to_status(status): 

68 """Convert a google.rpc.status.Status message to grpc.Status. 

69 

70 This is an EXPERIMENTAL API. 

71 

72 Args: 

73 status: a google.rpc.status.Status message representing the non-OK status 

74 to terminate the RPC with and communicate it to the client. 

75 

76 Returns: 

77 A grpc.Status instance representing the input google.rpc.status.Status message. 

78 """ 

79 return _Status( 

80 code=code_to_grpc_status_code(status.code), 

81 details=status.message, 

82 trailing_metadata=( 

83 (GRPC_DETAILS_METADATA_KEY, status.SerializeToString()), 

84 ), 

85 ) 

86 

87 

88__all__ = [ 

89 "from_call", 

90 "to_status", 

91] 

92 

93if sys.version_info[0] >= 3 and sys.version_info[1] >= 6: 

94 from . import _async as aio # pylint: disable=unused-import 

95 

96 __all__.append("aio")