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

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', 

28 ('code', 'details', 'trailing_metadata')), 

29 grpc.Status): 

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 if call.details() != rich_status.message: 

58 raise ValueError( 

59 'Message in Status proto (%s) doesn\'t match status details (%s)' 

60 % (rich_status.message, call.details())) 

61 return rich_status 

62 return None 

63 

64 

65def to_status(status): 

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

67 

68 This is an EXPERIMENTAL API. 

69 

70 Args: 

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

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

73 

74 Returns: 

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

76 """ 

77 return _Status(code=code_to_grpc_status_code(status.code), 

78 details=status.message, 

79 trailing_metadata=((GRPC_DETAILS_METADATA_KEY, 

80 status.SerializeToString()),)) 

81 

82 

83__all__ = [ 

84 'from_call', 

85 'to_status', 

86] 

87 

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

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

90 __all__.append('aio')