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
93
94
95
96
97
98
|
# Copyright 2016 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 reflection in gRPC Python."""
import sys
import grpc
from grpc_reflection.v1alpha import reflection_pb2 as _reflection_pb2
from grpc_reflection.v1alpha import reflection_pb2_grpc as _reflection_pb2_grpc
from grpc_reflection.v1alpha._base import BaseReflectionServicer
SERVICE_NAME = _reflection_pb2.DESCRIPTOR.services_by_name[
'ServerReflection'].full_name
class ReflectionServicer(BaseReflectionServicer):
"""Servicer handling RPCs for service statuses."""
def ServerReflectionInfo(self, request_iterator, context):
# pylint: disable=unused-argument
for request in request_iterator:
if request.HasField('file_by_filename'):
yield self._file_by_filename(request.file_by_filename)
elif request.HasField('file_containing_symbol'):
yield self._file_containing_symbol(
request.file_containing_symbol)
elif request.HasField('file_containing_extension'):
yield self._file_containing_extension(
request.file_containing_extension.containing_type,
request.file_containing_extension.extension_number)
elif request.HasField('all_extension_numbers_of_type'):
yield self._all_extension_numbers_of_type(
request.all_extension_numbers_of_type)
elif request.HasField('list_services'):
yield self._list_services()
else:
yield _reflection_pb2.ServerReflectionResponse(
error_response=_reflection_pb2.ErrorResponse(
error_code=grpc.StatusCode.INVALID_ARGUMENT.value[0],
error_message=grpc.StatusCode.INVALID_ARGUMENT.value[1].
encode(),
))
_enable_server_reflection_doc = """Enables server reflection on a server.
Args:
service_names: Iterable of fully-qualified service names available.
server: grpc.Server to which reflection service will be added.
pool: DescriptorPool object to use (descriptor_pool.Default() if None).
"""
if sys.version_info[0] >= 3 and sys.version_info[1] >= 6:
# Exposes AsyncReflectionServicer as public API.
from grpc.experimental import aio as grpc_aio # pylint: disable=ungrouped-imports
from . import _async as aio
def enable_server_reflection(service_names, server, pool=None):
if isinstance(server, grpc_aio.Server):
_reflection_pb2_grpc.add_ServerReflectionServicer_to_server(
aio.ReflectionServicer(service_names, pool=pool), server)
else:
_reflection_pb2_grpc.add_ServerReflectionServicer_to_server(
ReflectionServicer(service_names, pool=pool), server)
enable_server_reflection.__doc__ = _enable_server_reflection_doc
__all__ = [
"SERVICE_NAME",
"ReflectionServicer",
"enable_server_reflection",
"aio",
]
else:
def enable_server_reflection(service_names, server, pool=None):
_reflection_pb2_grpc.add_ServerReflectionServicer_to_server(
ReflectionServicer(service_names, pool=pool), server)
enable_server_reflection.__doc__ = _enable_server_reflection_doc
__all__ = [
"SERVICE_NAME",
"ReflectionServicer",
"enable_server_reflection",
]
|
