Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/aio/_server.py: 52%
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 # TODO(xuanwn): Implement this for AsyncIO.
99 pass
101 def add_insecure_port(self, address: str) -> int:
102 """Opens an insecure port for accepting RPCs.
104 This method may only be called before starting the server.
106 Args:
107 address: The address for which to open a port. If the port is 0,
108 or not specified in the address, then the gRPC runtime will choose a port.
110 Returns:
111 An integer port on which the server will accept RPC requests.
112 """
113 return _common.validate_port_binding_result(
114 address, self._server.add_insecure_port(_common.encode(address))
115 )
117 def add_secure_port(
118 self, address: str, server_credentials: grpc.ServerCredentials
119 ) -> int:
120 """Opens a secure port for accepting RPCs.
122 This method may only be called before starting the server.
124 Args:
125 address: The address for which to open a port.
126 if the port is 0, or not specified in the address, then the gRPC
127 runtime will choose a port.
128 server_credentials: A ServerCredentials object.
130 Returns:
131 An integer port on which the server will accept RPC requests.
132 """
133 return _common.validate_port_binding_result(
134 address,
135 self._server.add_secure_port(
136 _common.encode(address), server_credentials
137 ),
138 )
140 async def start(self) -> None:
141 """Starts this Server.
143 This method may only be called once. (i.e. it is not idempotent).
144 """
145 await self._server.start()
147 async def stop(self, grace: Optional[float]) -> None:
148 """Stops this Server.
150 This method immediately stops the server from servicing new RPCs in
151 all cases.
153 If a grace period is specified, this method waits until all active
154 RPCs are finished or until the grace period is reached. RPCs that haven't
155 been terminated within the grace period are aborted.
156 If a grace period is not specified (by passing None for grace), all
157 existing RPCs are aborted immediately and this method blocks until
158 the last RPC handler terminates.
160 This method is idempotent and may be called at any time. Passing a
161 smaller grace value in a subsequent call will have the effect of
162 stopping the Server sooner (passing None will have the effect of
163 stopping the server immediately). Passing a larger grace value in a
164 subsequent call will not have the effect of stopping the server later
165 (i.e. the most restrictive grace value is used).
167 Args:
168 grace: A duration of time in seconds or None.
169 """
170 await self._server.shutdown(grace)
172 async def wait_for_termination(
173 self, timeout: Optional[float] = None
174 ) -> bool:
175 """Block current coroutine until the server stops.
177 This is an EXPERIMENTAL API.
179 The wait will not consume computational resources during blocking, and
180 it will block until one of the two following conditions are met:
182 1) The server is stopped or terminated;
183 2) A timeout occurs if timeout is not `None`.
185 The timeout argument works in the same way as `threading.Event.wait()`.
186 https://docs.python.org/3/library/threading.html#threading.Event.wait
188 Args:
189 timeout: A floating point number specifying a timeout for the
190 operation in seconds.
192 Returns:
193 A bool indicates if the operation times out.
194 """
195 return await self._server.wait_for_termination(timeout)
197 def __del__(self):
198 """Schedules a graceful shutdown in current event loop.
200 The Cython AioServer doesn't hold a ref-count to this class. It should
201 be safe to slightly extend the underlying Cython object's life span.
202 """
203 if hasattr(self, "_server") and self._server.is_running():
204 cygrpc.schedule_coro_threadsafe(
205 self._server.shutdown(None),
206 self._loop,
207 )
210def server(
211 migration_thread_pool: Optional[Executor] = None,
212 handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None,
213 interceptors: Optional[Sequence[Any]] = None,
214 options: Optional[ChannelArgumentType] = None,
215 maximum_concurrent_rpcs: Optional[int] = None,
216 compression: Optional[grpc.Compression] = None,
217):
218 """Creates a Server with which RPCs can be serviced.
220 Args:
221 migration_thread_pool: A futures.ThreadPoolExecutor to be used by the
222 Server to execute non-AsyncIO RPC handlers for migration purpose.
223 handlers: An optional list of GenericRpcHandlers used for executing RPCs.
224 More handlers may be added by calling add_generic_rpc_handlers any time
225 before the server is started.
226 interceptors: An optional list of ServerInterceptor objects that observe
227 and optionally manipulate the incoming RPCs before handing them over to
228 handlers. The interceptors are given control in the order they are
229 specified. This is an EXPERIMENTAL API.
230 options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime)
231 to configure the channel.
232 maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
233 will service before returning RESOURCE_EXHAUSTED status, or None to
234 indicate no limit.
235 compression: An element of grpc.Compression, e.g.
236 grpc.Compression.Gzip. This compression algorithm will be used for the
237 lifetime of the server unless overridden by set_compression.
239 Returns:
240 A Server object.
241 """
242 return Server(
243 migration_thread_pool,
244 () if handlers is None else handlers,
245 () if interceptors is None else interceptors,
246 () if options is None else options,
247 maximum_concurrent_rpcs,
248 compression,
249 )