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._cython import cygrpc
24from . import _base_server
25from ._interceptor import ServerInterceptor
26from ._typing import ChannelArgumentType
29def _augment_channel_arguments(
30 base_options: ChannelArgumentType, compression: Optional[grpc.Compression]
31):
32 compression_option = _compression.create_channel_option(compression)
33 return tuple(base_options) + compression_option
36class Server(_base_server.Server):
37 """Serves RPCs."""
39 def __init__(
40 self,
41 thread_pool: Optional[Executor],
42 generic_handlers: Optional[Sequence[grpc.GenericRpcHandler]],
43 interceptors: Optional[Sequence[Any]],
44 options: ChannelArgumentType,
45 maximum_concurrent_rpcs: Optional[int],
46 compression: Optional[grpc.Compression],
47 ):
48 self._loop = cygrpc.get_working_loop()
49 if interceptors:
50 invalid_interceptors = [
51 interceptor
52 for interceptor in interceptors
53 if not isinstance(interceptor, ServerInterceptor)
54 ]
55 if invalid_interceptors:
56 error_msg = (
57 "Interceptor must be ServerInterceptor,"
58 "the following are invalid: {invalid_interceptors}"
59 )
60 # TODO(asheshvidyut): fix the value error below
61 # not caught by ruff.
62 raise ValueError(error_msg)
63 self._server = cygrpc.AioServer(
64 self._loop,
65 thread_pool,
66 generic_handlers,
67 interceptors,
68 _augment_channel_arguments(options, compression),
69 maximum_concurrent_rpcs,
70 )
72 def add_generic_rpc_handlers(
73 self, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]
74 ) -> None:
75 """Registers GenericRpcHandlers with this Server.
77 This method is only safe to call before the server is started.
79 Args:
80 generic_rpc_handlers: A sequence of GenericRpcHandlers that will be
81 used to service RPCs.
82 """
83 self._server.add_generic_rpc_handlers(generic_rpc_handlers)
85 def add_registered_method_handlers(
86 self,
87 service_name: str,
88 method_handlers: Dict[str, grpc.RpcMethodHandler],
89 ) -> None:
90 # TODO(xuanwn): Implement this for AsyncIO.
91 pass
93 def add_insecure_port(self, address: str) -> int:
94 """Opens an insecure port for accepting RPCs.
96 This method may only be called before starting the server.
98 Args:
99 address: The address for which to open a port. If the port is 0,
100 or not specified in the address, then the gRPC runtime will choose a port.
102 Returns:
103 An integer port on which the server will accept RPC requests.
104 """
105 return _common.validate_port_binding_result(
106 address, self._server.add_insecure_port(_common.encode(address))
107 )
109 def add_secure_port(
110 self, address: str, server_credentials: grpc.ServerCredentials
111 ) -> int:
112 """Opens a secure port for accepting RPCs.
114 This method may only be called before starting the server.
116 Args:
117 address: The address for which to open a port.
118 if the port is 0, or not specified in the address, then the gRPC
119 runtime will choose a port.
120 server_credentials: A ServerCredentials object.
122 Returns:
123 An integer port on which the server will accept RPC requests.
124 """
125 return _common.validate_port_binding_result(
126 address,
127 self._server.add_secure_port(
128 _common.encode(address), server_credentials
129 ),
130 )
132 async def start(self) -> None:
133 """Starts this Server.
135 This method may only be called once. (i.e. it is not idempotent).
136 """
137 await self._server.start()
139 async def stop(self, grace: Optional[float]) -> None:
140 """Stops this Server.
142 This method immediately stops the server from servicing new RPCs in
143 all cases.
145 If a grace period is specified, this method waits until all active
146 RPCs are finished or until the grace period is reached. RPCs that haven't
147 been terminated within the grace period are aborted.
148 If a grace period is not specified (by passing None for grace), all
149 existing RPCs are aborted immediately and this method blocks until
150 the last RPC handler terminates.
152 This method is idempotent and may be called at any time. Passing a
153 smaller grace value in a subsequent call will have the effect of
154 stopping the Server sooner (passing None will have the effect of
155 stopping the server immediately). Passing a larger grace value in a
156 subsequent call will not have the effect of stopping the server later
157 (i.e. the most restrictive grace value is used).
159 Args:
160 grace: A duration of time in seconds or None.
161 """
162 await self._server.shutdown(grace)
164 async def wait_for_termination(
165 self, timeout: Optional[float] = None
166 ) -> bool:
167 """Block current coroutine until the server stops.
169 This is an EXPERIMENTAL API.
171 The wait will not consume computational resources during blocking, and
172 it will block until one of the two following conditions are met:
174 1) The server is stopped or terminated;
175 2) A timeout occurs if timeout is not `None`.
177 The timeout argument works in the same way as `threading.Event.wait()`.
178 https://docs.python.org/3/library/threading.html#threading.Event.wait
180 Args:
181 timeout: A floating point number specifying a timeout for the
182 operation in seconds.
184 Returns:
185 A bool indicates if the operation times out.
186 """
187 return await self._server.wait_for_termination(timeout)
189 def __del__(self):
190 """Schedules a graceful shutdown in current event loop.
192 The Cython AioServer doesn't hold a ref-count to this class. It should
193 be safe to slightly extend the underlying Cython object's life span.
194 """
195 if hasattr(self, "_server") and self._server.is_running():
196 cygrpc.schedule_coro_threadsafe(
197 self._server.shutdown(None),
198 self._loop,
199 )
202def server(
203 migration_thread_pool: Optional[Executor] = None,
204 handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None,
205 interceptors: Optional[Sequence[Any]] = None,
206 options: Optional[ChannelArgumentType] = None,
207 maximum_concurrent_rpcs: Optional[int] = None,
208 compression: Optional[grpc.Compression] = None,
209):
210 """Creates a Server with which RPCs can be serviced.
212 Args:
213 migration_thread_pool: A futures.ThreadPoolExecutor to be used by the
214 Server to execute non-AsyncIO RPC handlers for migration purpose.
215 handlers: An optional list of GenericRpcHandlers used for executing RPCs.
216 More handlers may be added by calling add_generic_rpc_handlers any time
217 before the server is started.
218 interceptors: An optional list of ServerInterceptor objects that observe
219 and optionally manipulate the incoming RPCs before handing them over to
220 handlers. The interceptors are given control in the order they are
221 specified. This is an EXPERIMENTAL API.
222 options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime)
223 to configure the channel.
224 maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
225 will service before returning RESOURCE_EXHAUSTED status, or None to
226 indicate no limit.
227 compression: An element of grpc.compression, e.g.
228 grpc.compression.Gzip. This compression algorithm will be used for the
229 lifetime of the server unless overridden by set_compression.
231 Returns:
232 A Server object.
233 """
234 return Server(
235 migration_thread_pool,
236 () if handlers is None else handlers,
237 () if interceptors is None else interceptors,
238 () if options is None else options,
239 maximum_concurrent_rpcs,
240 compression,
241 )