1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
# Copyright 2018 The gRPC Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Reference implementation for status mapping in gRPC Python."""
import collections
import grpc
# TODO(https://github.com/bazelbuild/bazel/issues/6844)
# Due to Bazel issue, the namespace packages won't resolve correctly.
# Adding this unused-import as a workaround to avoid module-not-found error
# under Bazel builds.
import google.protobuf # pylint: disable=unused-import
from google.rpc import status_pb2
_CODE_TO_GRPC_CODE_MAPPING = dict([(x.value[0], x) for x in grpc.StatusCode])
_GRPC_DETAILS_METADATA_KEY = 'grpc-status-details-bin'
class _Status(
collections.namedtuple(
'_Status', ('code', 'details', 'trailing_metadata')), grpc.Status):
pass
def _code_to_grpc_status_code(code):
try:
return _CODE_TO_GRPC_CODE_MAPPING[code]
except KeyError:
raise ValueError('Invalid status code %s' % code)
def from_call(call):
"""Returns a google.rpc.status.Status message corresponding to a given grpc.Call.
This is an EXPERIMENTAL API.
Args:
call: A grpc.Call instance.
Returns:
A google.rpc.status.Status message representing the status of the RPC.
Raises:
ValueError: If the gRPC call's code or details are inconsistent with the
status code and message inside of the google.rpc.status.Status.
"""
for key, value in call.trailing_metadata():
if key == _GRPC_DETAILS_METADATA_KEY:
rich_status = status_pb2.Status.FromString(value)
if call.code().value[0] != rich_status.code:
raise ValueError(
'Code in Status proto (%s) doesn\'t match status code (%s)'
% (_code_to_grpc_status_code(rich_status.code),
call.code()))
if call.details() != rich_status.message:
raise ValueError(
'Message in Status proto (%s) doesn\'t match status details (%s)'
% (rich_status.message, call.details()))
return rich_status
return None
def to_status(status):
"""Convert a google.rpc.status.Status message to grpc.Status.
This is an EXPERIMENTAL API.
Args:
status: a google.rpc.status.Status message representing the non-OK status
to terminate the RPC with and communicate it to the client.
Returns:
A grpc.Status instance representing the input google.rpc.status.Status message.
"""
return _Status(
code=_code_to_grpc_status_code(status.code),
details=status.message,
trailing_metadata=((_GRPC_DETAILS_METADATA_KEY,
status.SerializeToString()),))
|
