1 # Copyright 2019 gRPC authors.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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 """Invocation-side implementation of gRPC Asyncio Python."""
18 from typing import Any, Iterable, Optional, Sequence, List
21 from grpc import _common, _compression, _grpcio_metadata
22 from grpc._cython import cygrpc
24 from . import _base_call, _base_channel
25 from ._call import (StreamStreamCall, StreamUnaryCall, UnaryStreamCall,
27 from ._interceptor import (
28 InterceptedUnaryUnaryCall, InterceptedUnaryStreamCall,
29 InterceptedStreamUnaryCall, InterceptedStreamStreamCall, ClientInterceptor,
30 UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
31 StreamUnaryClientInterceptor, StreamStreamClientInterceptor)
32 from ._metadata import Metadata
33 from ._typing import (ChannelArgumentType, DeserializingFunction,
34 SerializingFunction, RequestIterableType)
35 from ._utils import _timeout_to_deadline
37 _USER_AGENT = 'grpc-python-asyncio/{}'.format(_grpcio_metadata.__version__)
39 if sys.version_info[1] < 7:
41 def _all_tasks() -> Iterable[asyncio.Task]:
42 return asyncio.Task.all_tasks()
45 def _all_tasks() -> Iterable[asyncio.Task]:
46 return asyncio.all_tasks()
49 def _augment_channel_arguments(base_options: ChannelArgumentType,
50 compression: Optional[grpc.Compression]):
51 compression_channel_argument = _compression.create_channel_option(
53 user_agent_channel_argument = ((
54 cygrpc.ChannelArgKey.primary_user_agent_string,
57 return tuple(base_options
58 ) + compression_channel_argument + user_agent_channel_argument
61 class _BaseMultiCallable:
62 """Base class of all multi callable objects.
64 Handles the initialization logic and stores common attributes.
66 _loop: asyncio.AbstractEventLoop
67 _channel: cygrpc.AioChannel
69 _request_serializer: SerializingFunction
70 _response_deserializer: DeserializingFunction
71 _interceptors: Optional[Sequence[ClientInterceptor]]
72 _loop: asyncio.AbstractEventLoop
74 # pylint: disable=too-many-arguments
77 channel: cygrpc.AioChannel,
79 request_serializer: SerializingFunction,
80 response_deserializer: DeserializingFunction,
81 interceptors: Optional[Sequence[ClientInterceptor]],
82 loop: asyncio.AbstractEventLoop,
85 self._channel = channel
87 self._request_serializer = request_serializer
88 self._response_deserializer = response_deserializer
89 self._interceptors = interceptors
93 metadata: Optional[Metadata] = None,
94 compression: Optional[grpc.Compression] = None) -> Metadata:
95 """Based on the provided values for <metadata> or <compression> initialise the final
96 metadata, as it should be used for the current call.
98 metadata = metadata or Metadata()
101 *_compression.augment_metadata(metadata, compression))
105 class UnaryUnaryMultiCallable(_BaseMultiCallable,
106 _base_channel.UnaryUnaryMultiCallable):
112 timeout: Optional[float] = None,
113 metadata: Optional[Metadata] = None,
114 credentials: Optional[grpc.CallCredentials] = None,
115 wait_for_ready: Optional[bool] = None,
116 compression: Optional[grpc.Compression] = None
117 ) -> _base_call.UnaryUnaryCall:
119 metadata = self._init_metadata(metadata, compression)
120 if not self._interceptors:
121 call = UnaryUnaryCall(request, _timeout_to_deadline(timeout),
122 metadata, credentials, wait_for_ready,
123 self._channel, self._method,
124 self._request_serializer,
125 self._response_deserializer, self._loop)
127 call = InterceptedUnaryUnaryCall(
128 self._interceptors, request, timeout, metadata, credentials,
129 wait_for_ready, self._channel, self._method,
130 self._request_serializer, self._response_deserializer,
136 class UnaryStreamMultiCallable(_BaseMultiCallable,
137 _base_channel.UnaryStreamMultiCallable):
143 timeout: Optional[float] = None,
144 metadata: Optional[Metadata] = None,
145 credentials: Optional[grpc.CallCredentials] = None,
146 wait_for_ready: Optional[bool] = None,
147 compression: Optional[grpc.Compression] = None
148 ) -> _base_call.UnaryStreamCall:
150 metadata = self._init_metadata(metadata, compression)
151 deadline = _timeout_to_deadline(timeout)
153 if not self._interceptors:
154 call = UnaryStreamCall(request, deadline, metadata, credentials,
155 wait_for_ready, self._channel, self._method,
156 self._request_serializer,
157 self._response_deserializer, self._loop)
159 call = InterceptedUnaryStreamCall(
160 self._interceptors, request, deadline, metadata, credentials,
161 wait_for_ready, self._channel, self._method,
162 self._request_serializer, self._response_deserializer,
168 class StreamUnaryMultiCallable(_BaseMultiCallable,
169 _base_channel.StreamUnaryMultiCallable):
173 request_iterator: Optional[RequestIterableType] = None,
174 timeout: Optional[float] = None,
175 metadata: Optional[Metadata] = None,
176 credentials: Optional[grpc.CallCredentials] = None,
177 wait_for_ready: Optional[bool] = None,
178 compression: Optional[grpc.Compression] = None
179 ) -> _base_call.StreamUnaryCall:
181 metadata = self._init_metadata(metadata, compression)
182 deadline = _timeout_to_deadline(timeout)
184 if not self._interceptors:
185 call = StreamUnaryCall(request_iterator, deadline, metadata,
186 credentials, wait_for_ready, self._channel,
187 self._method, self._request_serializer,
188 self._response_deserializer, self._loop)
190 call = InterceptedStreamUnaryCall(
191 self._interceptors, request_iterator, deadline, metadata,
192 credentials, wait_for_ready, self._channel, self._method,
193 self._request_serializer, self._response_deserializer,
199 class StreamStreamMultiCallable(_BaseMultiCallable,
200 _base_channel.StreamStreamMultiCallable):
204 request_iterator: Optional[RequestIterableType] = None,
205 timeout: Optional[float] = None,
206 metadata: Optional[Metadata] = None,
207 credentials: Optional[grpc.CallCredentials] = None,
208 wait_for_ready: Optional[bool] = None,
209 compression: Optional[grpc.Compression] = None
210 ) -> _base_call.StreamStreamCall:
212 metadata = self._init_metadata(metadata, compression)
213 deadline = _timeout_to_deadline(timeout)
215 if not self._interceptors:
216 call = StreamStreamCall(request_iterator, deadline, metadata,
217 credentials, wait_for_ready, self._channel,
218 self._method, self._request_serializer,
219 self._response_deserializer, self._loop)
221 call = InterceptedStreamStreamCall(
222 self._interceptors, request_iterator, deadline, metadata,
223 credentials, wait_for_ready, self._channel, self._method,
224 self._request_serializer, self._response_deserializer,
230 class Channel(_base_channel.Channel):
231 _loop: asyncio.AbstractEventLoop
232 _channel: cygrpc.AioChannel
233 _unary_unary_interceptors: List[UnaryUnaryClientInterceptor]
234 _unary_stream_interceptors: List[UnaryStreamClientInterceptor]
235 _stream_unary_interceptors: List[StreamUnaryClientInterceptor]
236 _stream_stream_interceptors: List[StreamStreamClientInterceptor]
238 def __init__(self, target: str, options: ChannelArgumentType,
239 credentials: Optional[grpc.ChannelCredentials],
240 compression: Optional[grpc.Compression],
241 interceptors: Optional[Sequence[ClientInterceptor]]):
245 target: The target to which to connect.
246 options: Configuration options for the channel.
247 credentials: A cygrpc.ChannelCredentials or None.
248 compression: An optional value indicating the compression method to be
249 used over the lifetime of the channel.
250 interceptors: An optional list of interceptors that would be used for
251 intercepting any RPC executed with that channel.
253 self._unary_unary_interceptors = []
254 self._unary_stream_interceptors = []
255 self._stream_unary_interceptors = []
256 self._stream_stream_interceptors = []
258 if interceptors is not None:
259 for interceptor in interceptors:
260 if isinstance(interceptor, UnaryUnaryClientInterceptor):
261 self._unary_unary_interceptors.append(interceptor)
262 elif isinstance(interceptor, UnaryStreamClientInterceptor):
263 self._unary_stream_interceptors.append(interceptor)
264 elif isinstance(interceptor, StreamUnaryClientInterceptor):
265 self._stream_unary_interceptors.append(interceptor)
266 elif isinstance(interceptor, StreamStreamClientInterceptor):
267 self._stream_stream_interceptors.append(interceptor)
270 "Interceptor {} must be ".format(interceptor) +
271 "{} or ".format(UnaryUnaryClientInterceptor.__name__) +
272 "{} or ".format(UnaryStreamClientInterceptor.__name__) +
273 "{} or ".format(StreamUnaryClientInterceptor.__name__) +
274 "{}. ".format(StreamStreamClientInterceptor.__name__))
276 self._loop = cygrpc.get_working_loop()
277 self._channel = cygrpc.AioChannel(
278 _common.encode(target),
279 _augment_channel_arguments(options, compression), credentials,
282 async def __aenter__(self):
285 async def __aexit__(self, exc_type, exc_val, exc_tb):
286 await self._close(None)
288 async def _close(self, grace): # pylint: disable=too-many-branches
289 if self._channel.closed():
292 # No new calls will be accepted by the Cython channel.
293 self._channel.closing()
295 # Iterate through running tasks
301 stack = task.get_stack(limit=1)
302 except AttributeError as attribute_error:
303 # NOTE(lidiz) tl;dr: If the Task is created with a CPython
304 # object, it will trigger AttributeError.
306 # In the global finalizer, the event loop schedules
307 # a CPython PyAsyncGenAThrow object.
308 # https://github.com/python/cpython/blob/00e45877e33d32bb61aa13a2033e3bba370bda4d/Lib/asyncio/base_events.py#L484
310 # However, the PyAsyncGenAThrow object is written in C and
311 # failed to include the normal Python frame objects. Hence,
312 # this exception is a false negative, and it is safe to ignore
313 # the failure. It is fixed by https://github.com/python/cpython/pull/18669,
314 # but not available until 3.9 or 3.8.3. So, we have to keep it
316 # TODO(lidiz) drop this hack after 3.8 deprecation
317 if 'frame' in str(attribute_error):
322 # If the Task is created by a C-extension, the stack will be empty.
326 # Locate ones created by `aio.Call`.
328 candidate = frame.f_locals.get('self')
330 if isinstance(candidate, _base_call.Call):
331 if hasattr(candidate, '_channel'):
332 # For intercepted Call object
333 if candidate._channel is not self._channel:
335 elif hasattr(candidate, '_cython_call'):
336 # For normal Call object
337 if candidate._cython_call._channel is not self._channel:
340 # Unidentified Call object
341 raise cygrpc.InternalError(
342 f'Unrecognized call object: {candidate}')
344 calls.append(candidate)
345 call_tasks.append(task)
347 # If needed, try to wait for them to finish.
348 # Call objects are not always awaitables.
349 if grace and call_tasks:
350 await asyncio.wait(call_tasks, timeout=grace, loop=self._loop)
352 # Time to cancel existing calls.
356 # Destroy the channel
357 self._channel.close()
359 async def close(self, grace: Optional[float] = None):
360 await self._close(grace)
363 try_to_connect: bool = False) -> grpc.ChannelConnectivity:
364 result = self._channel.check_connectivity_state(try_to_connect)
365 return _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[result]
367 async def wait_for_state_change(
369 last_observed_state: grpc.ChannelConnectivity,
371 assert await self._channel.watch_connectivity_state(
372 last_observed_state.value[0], None)
374 async def channel_ready(self) -> None:
375 state = self.get_state(try_to_connect=True)
376 while state != grpc.ChannelConnectivity.READY:
377 await self.wait_for_state_change(state)
378 state = self.get_state(try_to_connect=True)
383 request_serializer: Optional[SerializingFunction] = None,
384 response_deserializer: Optional[DeserializingFunction] = None
385 ) -> UnaryUnaryMultiCallable:
386 return UnaryUnaryMultiCallable(self._channel, _common.encode(method),
388 response_deserializer,
389 self._unary_unary_interceptors,
395 request_serializer: Optional[SerializingFunction] = None,
396 response_deserializer: Optional[DeserializingFunction] = None
397 ) -> UnaryStreamMultiCallable:
398 return UnaryStreamMultiCallable(self._channel, _common.encode(method),
400 response_deserializer,
401 self._unary_stream_interceptors,
407 request_serializer: Optional[SerializingFunction] = None,
408 response_deserializer: Optional[DeserializingFunction] = None
409 ) -> StreamUnaryMultiCallable:
410 return StreamUnaryMultiCallable(self._channel, _common.encode(method),
412 response_deserializer,
413 self._stream_unary_interceptors,
419 request_serializer: Optional[SerializingFunction] = None,
420 response_deserializer: Optional[DeserializingFunction] = None
421 ) -> StreamStreamMultiCallable:
422 return StreamStreamMultiCallable(self._channel, _common.encode(method),
424 response_deserializer,
425 self._stream_stream_interceptors,
429 def insecure_channel(
431 options: Optional[ChannelArgumentType] = None,
432 compression: Optional[grpc.Compression] = None,
433 interceptors: Optional[Sequence[ClientInterceptor]] = None):
434 """Creates an insecure asynchronous Channel to a server.
437 target: The server address
438 options: An optional list of key-value pairs (:term:`channel_arguments`
439 in gRPC Core runtime) to configure the channel.
440 compression: An optional value indicating the compression method to be
441 used over the lifetime of the channel. This is an EXPERIMENTAL option.
442 interceptors: An optional sequence of interceptors that will be executed for
443 any call executed with this channel.
448 return Channel(target, () if options is None else options, None,
449 compression, interceptors)
452 def secure_channel(target: str,
453 credentials: grpc.ChannelCredentials,
454 options: Optional[ChannelArgumentType] = None,
455 compression: Optional[grpc.Compression] = None,
456 interceptors: Optional[Sequence[ClientInterceptor]] = None):
457 """Creates a secure asynchronous Channel to a server.
460 target: The server address.
461 credentials: A ChannelCredentials instance.
462 options: An optional list of key-value pairs (:term:`channel_arguments`
463 in gRPC Core runtime) to configure the channel.
464 compression: An optional value indicating the compression method to be
465 used over the lifetime of the channel. This is an EXPERIMENTAL option.
466 interceptors: An optional sequence of interceptors that will be executed for
467 any call executed with this channel.
472 return Channel(target, () if options is None else options,
473 credentials._credentials, compression, interceptors)