Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/grpc/aio/_server.py: 54%
39 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 06:37 +0000
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, 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 raise ValueError(
57 "Interceptor must be ServerInterceptor, the "
58 f"following are invalid: {invalid_interceptors}"
59 )
60 self._server = cygrpc.AioServer(
61 self._loop,
62 thread_pool,
63 generic_handlers,
64 interceptors,
65 _augment_channel_arguments(options, compression),
66 maximum_concurrent_rpcs,
67 )
69 def add_generic_rpc_handlers(
70 self, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]
71 ) -> None:
72 """Registers GenericRpcHandlers with this Server.
74 This method is only safe to call before the server is started.
76 Args:
77 generic_rpc_handlers: A sequence of GenericRpcHandlers that will be
78 used to service RPCs.
79 """
80 self._server.add_generic_rpc_handlers(generic_rpc_handlers)
82 def add_insecure_port(self, address: str) -> int:
83 """Opens an insecure port for accepting RPCs.
85 This method may only be called before starting the server.
87 Args:
88 address: The address for which to open a port. If the port is 0,
89 or not specified in the address, then the gRPC runtime will choose a port.
91 Returns:
92 An integer port on which the server will accept RPC requests.
93 """
94 return _common.validate_port_binding_result(
95 address, self._server.add_insecure_port(_common.encode(address))
96 )
98 def add_secure_port(
99 self, address: str, server_credentials: grpc.ServerCredentials
100 ) -> int:
101 """Opens a secure port for accepting RPCs.
103 This method may only be called before starting the server.
105 Args:
106 address: The address for which to open a port.
107 if the port is 0, or not specified in the address, then the gRPC
108 runtime will choose a port.
109 server_credentials: A ServerCredentials object.
111 Returns:
112 An integer port on which the server will accept RPC requests.
113 """
114 return _common.validate_port_binding_result(
115 address,
116 self._server.add_secure_port(
117 _common.encode(address), server_credentials
118 ),
119 )
121 async def start(self) -> None:
122 """Starts this Server.
124 This method may only be called once. (i.e. it is not idempotent).
125 """
126 await self._server.start()
128 async def stop(self, grace: Optional[float]) -> None:
129 """Stops this Server.
131 This method immediately stops the server from servicing new RPCs in
132 all cases.
134 If a grace period is specified, this method returns immediately and all
135 RPCs active at the end of the grace period are aborted. If a grace
136 period is not specified (by passing None for grace), all existing RPCs
137 are aborted immediately and this method blocks until the last RPC
138 handler terminates.
140 This method is idempotent and may be called at any time. Passing a
141 smaller grace value in a subsequent call will have the effect of
142 stopping the Server sooner (passing None will have the effect of
143 stopping the server immediately). Passing a larger grace value in a
144 subsequent call will not have the effect of stopping the server later
145 (i.e. the most restrictive grace value is used).
147 Args:
148 grace: A duration of time in seconds or None.
149 """
150 await self._server.shutdown(grace)
152 async def wait_for_termination(
153 self, timeout: Optional[float] = None
154 ) -> bool:
155 """Block current coroutine until the server stops.
157 This is an EXPERIMENTAL API.
159 The wait will not consume computational resources during blocking, and
160 it will block until one of the two following conditions are met:
162 1) The server is stopped or terminated;
163 2) A timeout occurs if timeout is not `None`.
165 The timeout argument works in the same way as `threading.Event.wait()`.
166 https://docs.python.org/3/library/threading.html#threading.Event.wait
168 Args:
169 timeout: A floating point number specifying a timeout for the
170 operation in seconds.
172 Returns:
173 A bool indicates if the operation times out.
174 """
175 return await self._server.wait_for_termination(timeout)
177 def __del__(self):
178 """Schedules a graceful shutdown in current event loop.
180 The Cython AioServer doesn't hold a ref-count to this class. It should
181 be safe to slightly extend the underlying Cython object's life span.
182 """
183 if hasattr(self, "_server"):
184 if self._server.is_running():
185 cygrpc.schedule_coro_threadsafe(
186 self._server.shutdown(None),
187 self._loop,
188 )
191def server(
192 migration_thread_pool: Optional[Executor] = None,
193 handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None,
194 interceptors: Optional[Sequence[Any]] = None,
195 options: Optional[ChannelArgumentType] = None,
196 maximum_concurrent_rpcs: Optional[int] = None,
197 compression: Optional[grpc.Compression] = None,
198):
199 """Creates a Server with which RPCs can be serviced.
201 Args:
202 migration_thread_pool: A futures.ThreadPoolExecutor to be used by the
203 Server to execute non-AsyncIO RPC handlers for migration purpose.
204 handlers: An optional list of GenericRpcHandlers used for executing RPCs.
205 More handlers may be added by calling add_generic_rpc_handlers any time
206 before the server is started.
207 interceptors: An optional list of ServerInterceptor objects that observe
208 and optionally manipulate the incoming RPCs before handing them over to
209 handlers. The interceptors are given control in the order they are
210 specified. This is an EXPERIMENTAL API.
211 options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime)
212 to configure the channel.
213 maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
214 will service before returning RESOURCE_EXHAUSTED status, or None to
215 indicate no limit.
216 compression: An element of grpc.compression, e.g.
217 grpc.compression.Gzip. This compression algorithm will be used for the
218 lifetime of the server unless overridden by set_compression.
220 Returns:
221 A Server object.
222 """
223 return Server(
224 migration_thread_pool,
225 () if handlers is None else handlers,
226 () if interceptors is None else interceptors,
227 () if options is None else options,
228 maximum_concurrent_rpcs,
229 compression,
230 )