Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/aio/_server.py: 51%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2019 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"""Server-side implementation of gRPC Asyncio Python."""
16from concurrent.futures import Executor
17from typing import Any, Dict, Optional, Sequence
19import grpc
20from grpc import _common
21from grpc import _compression
22from grpc import _observability
23from grpc._cython import cygrpc
25from . import _base_server
26from ._interceptor import ServerInterceptor
27from ._typing import ChannelArgumentType
30def _augment_channel_arguments(
31 base_options: ChannelArgumentType, compression: Optional[grpc.Compression]
32):
33 compression_option = _compression.create_channel_option(compression)
34 maybe_server_call_tracer_factory_option = (
35 _observability.create_server_call_tracer_factory_option(xds=False)
36 )
37 return (
38 tuple(base_options)
39 + compression_option
40 + maybe_server_call_tracer_factory_option
41 )
44class Server(_base_server.Server):
45 """Serves RPCs."""
47 def __init__(
48 self,
49 thread_pool: Optional[Executor],
50 generic_handlers: Optional[Sequence[grpc.GenericRpcHandler]],
51 interceptors: Optional[Sequence[Any]],
52 options: ChannelArgumentType,
53 maximum_concurrent_rpcs: Optional[int],
54 compression: Optional[grpc.Compression],
55 ):
56 self._loop = cygrpc.get_working_loop()
57 if interceptors:
58 invalid_interceptors = [
59 interceptor
60 for interceptor in interceptors
61 if not isinstance(interceptor, ServerInterceptor)
62 ]
63 if invalid_interceptors:
64 error_msg = (
65 "Interceptor must be ServerInterceptor,"
66 "the following are invalid: {invalid_interceptors}"
67 )
68 # TODO(asheshvidyut): fix the value error below
69 # not caught by ruff.
70 raise ValueError(error_msg)
71 self._server = cygrpc.AioServer(
72 self._loop,
73 thread_pool,
74 generic_handlers,
75 interceptors,
76 _augment_channel_arguments(options, compression),
77 maximum_concurrent_rpcs,
78 )
80 def add_generic_rpc_handlers(
81 self, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]
82 ) -> None:
83 """Registers GenericRpcHandlers with this Server.
85 This method is only safe to call before the server is started.
87 Args:
88 generic_rpc_handlers: A sequence of GenericRpcHandlers that will be
89 used to service RPCs.
90 """
91 self._server.add_generic_rpc_handlers(generic_rpc_handlers)
93 def add_registered_method_handlers(
94 self,
95 service_name: str,
96 method_handlers: Dict[str, grpc.RpcMethodHandler],
97 ) -> None:
98 """Registers method handlers for pre-registered methods.
100 This method is only safe to call before server is started.
102 If the same method has both generic and registered handler,
103 registered handler will take precedence.
105 Args:
106 service_name: The service name.
107 method_handlers: A dictionary that maps method names to corresponding
108 RpcMethodHandler.
109 """
110 method_to_handlers = {
111 _common.fully_qualified_method(service_name, method): method_handler
112 for method, method_handler in method_handlers.items()
113 }
114 self._server.add_registered_method_handlers(method_to_handlers)
116 def add_insecure_port(self, address: str) -> int:
117 """Opens an insecure port for accepting RPCs.
119 This method may only be called before starting the server.
121 Args:
122 address: The address for which to open a port. If the port is 0,
123 or not specified in the address, then the gRPC runtime will choose a port.
125 Returns:
126 An integer port on which the server will accept RPC requests.
127 """
128 return _common.validate_port_binding_result(
129 address, self._server.add_insecure_port(_common.encode(address))
130 )
132 def add_secure_port(
133 self, address: str, server_credentials: grpc.ServerCredentials
134 ) -> int:
135 """Opens a secure port for accepting RPCs.
137 This method may only be called before starting the server.
139 Args:
140 address: The address for which to open a port.
141 if the port is 0, or not specified in the address, then the gRPC
142 runtime will choose a port.
143 server_credentials: A ServerCredentials object.
145 Returns:
146 An integer port on which the server will accept RPC requests.
147 """
148 return _common.validate_port_binding_result(
149 address,
150 self._server.add_secure_port(
151 _common.encode(address), server_credentials
152 ),
153 )
155 async def start(self) -> None:
156 """Starts this Server.
158 This method may only be called once. (i.e. it is not idempotent).
159 """
160 await self._server.start()
162 async def stop(self, grace: Optional[float]) -> None:
163 """Stops this Server.
165 This method immediately stops the server from servicing new RPCs in
166 all cases.
168 If a grace period is specified, this method waits until all active
169 RPCs are finished or until the grace period is reached. RPCs that haven't
170 been terminated within the grace period are aborted.
171 If a grace period is not specified (by passing None for grace), all
172 existing RPCs are aborted immediately and this method blocks until
173 the last RPC handler terminates.
175 This method is idempotent and may be called at any time. Passing a
176 smaller grace value in a subsequent call will have the effect of
177 stopping the Server sooner (passing None will have the effect of
178 stopping the server immediately). Passing a larger grace value in a
179 subsequent call will not have the effect of stopping the server later
180 (i.e. the most restrictive grace value is used).
182 Args:
183 grace: A duration of time in seconds or None.
184 """
185 await self._server.shutdown(grace)
187 async def wait_for_termination(
188 self, timeout: Optional[float] = None
189 ) -> bool:
190 """Block current coroutine until the server stops.
192 This is an EXPERIMENTAL API.
194 The wait will not consume computational resources during blocking, and
195 it will block until one of the two following conditions are met:
197 1) The server is stopped or terminated;
198 2) A timeout occurs if timeout is not `None`.
200 The timeout argument works in the same way as `threading.Event.wait()`.
201 https://docs.python.org/3/library/threading.html#threading.Event.wait
203 Args:
204 timeout: A floating point number specifying a timeout for the
205 operation in seconds.
207 Returns:
208 A bool indicates if the operation times out.
209 """
210 return await self._server.wait_for_termination(timeout)
212 def __del__(self):
213 """Schedules a graceful shutdown in current event loop.
215 The Cython AioServer doesn't hold a ref-count to this class. It should
216 be safe to slightly extend the underlying Cython object's life span.
217 """
218 if hasattr(self, "_server") and self._server.is_running():
219 cygrpc.schedule_coro_threadsafe(
220 self._server.shutdown(None),
221 self._loop,
222 )
225def server(
226 migration_thread_pool: Optional[Executor] = None,
227 handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None,
228 interceptors: Optional[Sequence[Any]] = None,
229 options: Optional[ChannelArgumentType] = None,
230 maximum_concurrent_rpcs: Optional[int] = None,
231 compression: Optional[grpc.Compression] = None,
232):
233 """Creates a Server with which RPCs can be serviced.
235 Args:
236 migration_thread_pool: A futures.ThreadPoolExecutor to be used by the
237 Server to execute non-AsyncIO RPC handlers for migration purpose.
238 handlers: An optional list of GenericRpcHandlers used for executing RPCs.
239 More handlers may be added by calling add_generic_rpc_handlers any time
240 before the server is started.
241 interceptors: An optional list of ServerInterceptor objects that observe
242 and optionally manipulate the incoming RPCs before handing them over to
243 handlers. The interceptors are given control in the order they are
244 specified. This is an EXPERIMENTAL API.
245 options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime)
246 to configure the channel.
247 maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
248 will service before returning RESOURCE_EXHAUSTED status, or None to
249 indicate no limit.
250 compression: An element of grpc.Compression, e.g.
251 grpc.Compression.Gzip. This compression algorithm will be used for the
252 lifetime of the server unless overridden by set_compression.
254 Returns:
255 A Server object.
256 """
257 return Server(
258 migration_thread_pool,
259 () if handlers is None else handlers,
260 () if interceptors is None else interceptors,
261 () if options is None else options,
262 maximum_concurrent_rpcs,
263 compression,
264 )