Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/aio/_channel.py: 42%
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 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"""Invocation-side implementation of gRPC Asyncio Python."""
16import asyncio
17from typing import Any, List, Optional, Sequence
18import weakref
20import grpc
21from grpc import _common
22from grpc import _compression
23from grpc import _grpcio_metadata
24from grpc._cython import cygrpc
26from . import _base_call
27from . import _base_channel
28from ._call import StreamStreamCall
29from ._call import StreamUnaryCall
30from ._call import UnaryStreamCall
31from ._call import UnaryUnaryCall
32from ._interceptor import ClientInterceptor
33from ._interceptor import InterceptedStreamStreamCall
34from ._interceptor import InterceptedStreamUnaryCall
35from ._interceptor import InterceptedUnaryStreamCall
36from ._interceptor import InterceptedUnaryUnaryCall
37from ._interceptor import StreamStreamClientInterceptor
38from ._interceptor import StreamUnaryClientInterceptor
39from ._interceptor import UnaryStreamClientInterceptor
40from ._interceptor import UnaryUnaryClientInterceptor
41from ._metadata import Metadata
42from ._typing import ChannelArgumentType
43from ._typing import DeserializingFunction
44from ._typing import MetadataType
45from ._typing import RequestIterableType
46from ._typing import RequestType
47from ._typing import ResponseType
48from ._typing import SerializingFunction
49from ._utils import _timeout_to_deadline
51_USER_AGENT = "grpc-python-asyncio/{}".format(_grpcio_metadata.__version__)
54def _augment_channel_arguments(
55 base_options: ChannelArgumentType, compression: Optional[grpc.Compression]
56):
57 compression_channel_argument = _compression.create_channel_option(
58 compression
59 )
60 user_agent_channel_argument = (
61 (
62 cygrpc.ChannelArgKey.primary_user_agent_string,
63 _USER_AGENT,
64 ),
65 )
66 return (
67 tuple(base_options)
68 + compression_channel_argument
69 + user_agent_channel_argument
70 )
73class _BaseMultiCallable:
74 """Base class of all multi callable objects.
76 Handles the initialization logic and stores common attributes.
77 """
79 _loop: asyncio.AbstractEventLoop
80 _channel: cygrpc.AioChannel
81 _method: bytes
82 _request_serializer: Optional[SerializingFunction]
83 _response_deserializer: Optional[DeserializingFunction]
84 _interceptors: Optional[Sequence[ClientInterceptor]]
85 _references: List[Any]
86 _loop: asyncio.AbstractEventLoop
88 # pylint: disable=too-many-arguments
89 def __init__(
90 self,
91 channel: cygrpc.AioChannel,
92 method: bytes,
93 request_serializer: Optional[SerializingFunction],
94 response_deserializer: Optional[DeserializingFunction],
95 interceptors: Optional[Sequence[ClientInterceptor]],
96 references: List[Any],
97 loop: asyncio.AbstractEventLoop,
98 ) -> None:
99 self._loop = loop
100 self._channel = channel
101 self._method = method
102 self._request_serializer = request_serializer
103 self._response_deserializer = response_deserializer
104 self._interceptors = interceptors
105 self._references = references
107 if not self._references:
108 error_msg = (
109 "MultiCallable must be attached to a Channel, unexpectedly"
110 " found no references."
111 )
112 raise ValueError(error_msg)
113 if not isinstance(self._references[0], Channel):
114 error_msg = (
115 "Invalid reference type. MultiCallable must be attached to a"
116 " Channel."
117 )
118 raise TypeError(error_msg)
120 self._python_channel = self._references[0]
122 @staticmethod
123 def _init_metadata(
124 metadata: Optional[MetadataType] = None,
125 compression: Optional[grpc.Compression] = None,
126 ) -> Metadata:
127 """Based on the provided values for <metadata> or <compression> initialise the final
128 metadata, as it should be used for the current call.
129 """
130 metadata = metadata or Metadata()
131 if not isinstance(metadata, Metadata) and isinstance(
132 metadata, Sequence
133 ):
134 metadata = Metadata.from_tuple(tuple(metadata))
135 if compression:
136 metadata = Metadata(
137 *_compression.augment_metadata(metadata, compression)
138 )
139 return metadata
142class UnaryUnaryMultiCallable(
143 _BaseMultiCallable, _base_channel.UnaryUnaryMultiCallable
144):
145 def __call__(
146 self,
147 request: RequestType,
148 *,
149 timeout: Optional[float] = None,
150 metadata: Optional[MetadataType] = None,
151 credentials: Optional[grpc.CallCredentials] = None,
152 wait_for_ready: Optional[bool] = None,
153 compression: Optional[grpc.Compression] = None,
154 ) -> _base_call.UnaryUnaryCall[RequestType, ResponseType]:
155 metadata = self._init_metadata(metadata, compression)
156 if not self._interceptors:
157 call = UnaryUnaryCall(
158 request,
159 _timeout_to_deadline(timeout),
160 metadata,
161 credentials,
162 wait_for_ready,
163 self._channel,
164 self._method,
165 self._request_serializer,
166 self._response_deserializer,
167 self._loop,
168 )
169 else:
170 call = InterceptedUnaryUnaryCall(
171 self._interceptors,
172 request,
173 timeout,
174 metadata,
175 credentials,
176 wait_for_ready,
177 self._channel,
178 self._method,
179 self._request_serializer,
180 self._response_deserializer,
181 self._loop,
182 )
184 self._python_channel._register_call(call)
186 return call
189class UnaryStreamMultiCallable(
190 _BaseMultiCallable, _base_channel.UnaryStreamMultiCallable
191):
192 def __call__(
193 self,
194 request: RequestType,
195 *,
196 timeout: Optional[float] = None,
197 metadata: Optional[MetadataType] = None,
198 credentials: Optional[grpc.CallCredentials] = None,
199 wait_for_ready: Optional[bool] = None,
200 compression: Optional[grpc.Compression] = None,
201 ) -> _base_call.UnaryStreamCall[RequestType, ResponseType]:
202 metadata = self._init_metadata(metadata, compression)
204 if not self._interceptors:
205 call = UnaryStreamCall(
206 request,
207 _timeout_to_deadline(timeout),
208 metadata,
209 credentials,
210 wait_for_ready,
211 self._channel,
212 self._method,
213 self._request_serializer,
214 self._response_deserializer,
215 self._loop,
216 )
217 else:
218 call = InterceptedUnaryStreamCall(
219 self._interceptors,
220 request,
221 timeout,
222 metadata,
223 credentials,
224 wait_for_ready,
225 self._channel,
226 self._method,
227 self._request_serializer,
228 self._response_deserializer,
229 self._loop,
230 )
232 self._python_channel._register_call(call)
234 return call
237class StreamUnaryMultiCallable(
238 _BaseMultiCallable, _base_channel.StreamUnaryMultiCallable
239):
240 def __call__(
241 self,
242 request_iterator: Optional[RequestIterableType] = None,
243 timeout: Optional[float] = None,
244 metadata: Optional[MetadataType] = None,
245 credentials: Optional[grpc.CallCredentials] = None,
246 wait_for_ready: Optional[bool] = None,
247 compression: Optional[grpc.Compression] = None,
248 ) -> _base_call.StreamUnaryCall:
249 metadata = self._init_metadata(metadata, compression)
251 if not self._interceptors:
252 call = StreamUnaryCall(
253 request_iterator,
254 _timeout_to_deadline(timeout),
255 metadata,
256 credentials,
257 wait_for_ready,
258 self._channel,
259 self._method,
260 self._request_serializer,
261 self._response_deserializer,
262 self._loop,
263 )
264 else:
265 call = InterceptedStreamUnaryCall(
266 self._interceptors,
267 request_iterator,
268 timeout,
269 metadata,
270 credentials,
271 wait_for_ready,
272 self._channel,
273 self._method,
274 self._request_serializer,
275 self._response_deserializer,
276 self._loop,
277 )
279 self._python_channel._register_call(call)
281 return call
284class StreamStreamMultiCallable(
285 _BaseMultiCallable, _base_channel.StreamStreamMultiCallable
286):
287 def __call__(
288 self,
289 request_iterator: Optional[RequestIterableType] = None,
290 timeout: Optional[float] = None,
291 metadata: Optional[MetadataType] = None,
292 credentials: Optional[grpc.CallCredentials] = None,
293 wait_for_ready: Optional[bool] = None,
294 compression: Optional[grpc.Compression] = None,
295 ) -> _base_call.StreamStreamCall:
296 metadata = self._init_metadata(metadata, compression)
298 if not self._interceptors:
299 call = StreamStreamCall(
300 request_iterator,
301 _timeout_to_deadline(timeout),
302 metadata,
303 credentials,
304 wait_for_ready,
305 self._channel,
306 self._method,
307 self._request_serializer,
308 self._response_deserializer,
309 self._loop,
310 )
311 else:
312 call = InterceptedStreamStreamCall(
313 self._interceptors,
314 request_iterator,
315 timeout,
316 metadata,
317 credentials,
318 wait_for_ready,
319 self._channel,
320 self._method,
321 self._request_serializer,
322 self._response_deserializer,
323 self._loop,
324 )
326 self._python_channel._register_call(call)
328 return call
331class Channel(_base_channel.Channel):
332 _loop: asyncio.AbstractEventLoop
333 _channel: cygrpc.AioChannel
334 _unary_unary_interceptors: List[UnaryUnaryClientInterceptor]
335 _unary_stream_interceptors: List[UnaryStreamClientInterceptor]
336 _stream_unary_interceptors: List[StreamUnaryClientInterceptor]
337 _stream_stream_interceptors: List[StreamStreamClientInterceptor]
339 def __init__(
340 self,
341 target: str,
342 options: ChannelArgumentType,
343 credentials: Optional[cygrpc.ChannelCredentials],
344 compression: Optional[grpc.Compression],
345 interceptors: Optional[Sequence[ClientInterceptor]],
346 ):
347 """Constructor.
349 Args:
350 target: The target to which to connect.
351 options: Configuration options for the channel.
352 credentials: A cygrpc.ChannelCredentials or None.
353 compression: An optional value indicating the compression method to be
354 used over the lifetime of the channel.
355 interceptors: An optional list of interceptors that would be used for
356 intercepting any RPC executed with that channel.
357 """
358 self._unary_unary_interceptors = []
359 self._unary_stream_interceptors = []
360 self._stream_unary_interceptors = []
361 self._stream_stream_interceptors = []
363 if interceptors is not None:
364 for interceptor in interceptors:
365 if isinstance(interceptor, UnaryUnaryClientInterceptor):
366 self._unary_unary_interceptors.append(interceptor)
367 elif isinstance(interceptor, UnaryStreamClientInterceptor):
368 self._unary_stream_interceptors.append(interceptor)
369 elif isinstance(interceptor, StreamUnaryClientInterceptor):
370 self._stream_unary_interceptors.append(interceptor)
371 elif isinstance(interceptor, StreamStreamClientInterceptor):
372 self._stream_stream_interceptors.append(interceptor)
373 else:
374 raise ValueError( # noqa: TRY004
375 "Interceptor {} must be ".format(interceptor)
376 + "{} or ".format(UnaryUnaryClientInterceptor.__name__)
377 + "{} or ".format(UnaryStreamClientInterceptor.__name__)
378 + "{} or ".format(StreamUnaryClientInterceptor.__name__)
379 + "{}. ".format(StreamStreamClientInterceptor.__name__)
380 )
382 self._loop = cygrpc.get_working_loop()
383 self._channel = cygrpc.AioChannel(
384 _common.encode(target),
385 _augment_channel_arguments(options, compression),
386 credentials,
387 self._loop,
388 )
389 self._active_calls = weakref.WeakSet()
391 def _register_call(self, call: _base_call.Call) -> None:
392 """Register a call to be tracked by the channel."""
393 self._active_calls.add(call)
394 call.add_done_callback(self._active_calls.discard)
396 async def __aenter__(self):
397 return self
399 async def __aexit__(self, exc_type, exc_val, exc_tb):
400 await self._close(None)
402 async def _close(self, grace): # pylint: disable=too-many-branches
403 if self._channel.closed():
404 return
406 if grace and grace < 0:
407 error_msg = f"grace must be non-negative, got {grace}."
408 raise ValueError(error_msg)
410 # No new calls will be accepted by the Cython channel.
411 self._channel.closing()
413 async def _wait_for_call_to_complete(call):
414 try:
415 await call.code()
416 except Exception: # pylint: disable=broad-except
417 # Ignore exceptions here as true RPC errors bubble up via
418 # standard application paths. Silencing prevents channel close
419 # from failing and suppresses asyncio noise warnings.
420 pass
422 calls = list(self._active_calls)
424 if grace:
425 call_tasks = [
426 self._loop.create_task(_wait_for_call_to_complete(call))
427 for call in calls
428 if not call.done()
429 ]
430 if call_tasks:
431 await asyncio.wait(call_tasks, timeout=grace)
433 # Time to cancel existing calls.
434 for call in calls:
435 call.cancel()
437 calls.clear()
438 self._active_calls.clear()
440 # Destroy the channel
441 self._channel.close()
443 async def close(self, grace: Optional[float] = None):
444 await self._close(grace)
446 def __del__(self):
447 if hasattr(self, "_channel") and not self._channel.closed():
448 self._channel.close()
450 def get_state(
451 self, try_to_connect: bool = False
452 ) -> grpc.ChannelConnectivity:
453 result = self._channel.check_connectivity_state(try_to_connect)
454 return _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[result]
456 async def wait_for_state_change(
457 self,
458 last_observed_state: grpc.ChannelConnectivity,
459 ) -> None:
460 # We raise a RuntimeError if watch_connectivity_state returns False.
461 #
462 # The watch_connectivity_state method returns True when it observes a state change
463 # and False when it times out (which shouldn't happen since no timeout is specified).
464 # A channel close triggers a transition to SHUTDOWN, which resolves all pending watch
465 # calls and makes them return True. Thus, watch_connectivity_state should only return
466 # True under normal operation; returning False indicates an implementation issue.
467 #
468 # We do not use an assert statement here because asserts
469 # can be optimized out under python -O.
470 # See https://github.com/grpc/grpc/issues/42393 for context.
471 resolved = await self._channel.watch_connectivity_state(
472 last_observed_state.value[0], None
473 )
474 if not resolved:
475 error_msg = (
476 "gRPC channel connectivity state watch failed unexpectedly."
477 )
478 raise RuntimeError(error_msg)
480 async def channel_ready(self) -> None:
481 state = self.get_state(try_to_connect=True)
482 while state != grpc.ChannelConnectivity.READY:
483 await self.wait_for_state_change(state)
484 state = self.get_state(try_to_connect=True)
486 # TODO(xuanwn): Implement this method after we have
487 # observability for Asyncio.
488 def _get_registered_call_handle(self, method: str) -> int:
489 pass
491 # TODO(xuanwn): Implement _registered_method after we have
492 # observability for Asyncio.
493 # pylint: disable=arguments-differ,unused-argument
494 def unary_unary(
495 self,
496 method: str,
497 request_serializer: Optional[SerializingFunction] = None,
498 response_deserializer: Optional[DeserializingFunction] = None,
499 _registered_method: Optional[bool] = False,
500 ) -> UnaryUnaryMultiCallable:
501 return UnaryUnaryMultiCallable(
502 self._channel,
503 _common.encode(method),
504 request_serializer,
505 response_deserializer,
506 self._unary_unary_interceptors,
507 [self],
508 self._loop,
509 )
511 # TODO(xuanwn): Implement _registered_method after we have
512 # observability for Asyncio.
513 # pylint: disable=arguments-differ,unused-argument
514 def unary_stream(
515 self,
516 method: str,
517 request_serializer: Optional[SerializingFunction] = None,
518 response_deserializer: Optional[DeserializingFunction] = None,
519 _registered_method: Optional[bool] = False,
520 ) -> UnaryStreamMultiCallable:
521 return UnaryStreamMultiCallable(
522 self._channel,
523 _common.encode(method),
524 request_serializer,
525 response_deserializer,
526 self._unary_stream_interceptors,
527 [self],
528 self._loop,
529 )
531 # TODO(xuanwn): Implement _registered_method after we have
532 # observability for Asyncio.
533 # pylint: disable=arguments-differ,unused-argument
534 def stream_unary(
535 self,
536 method: str,
537 request_serializer: Optional[SerializingFunction] = None,
538 response_deserializer: Optional[DeserializingFunction] = None,
539 _registered_method: Optional[bool] = False,
540 ) -> StreamUnaryMultiCallable:
541 return StreamUnaryMultiCallable(
542 self._channel,
543 _common.encode(method),
544 request_serializer,
545 response_deserializer,
546 self._stream_unary_interceptors,
547 [self],
548 self._loop,
549 )
551 # TODO(xuanwn): Implement _registered_method after we have
552 # observability for Asyncio.
553 # pylint: disable=arguments-differ,unused-argument
554 def stream_stream(
555 self,
556 method: str,
557 request_serializer: Optional[SerializingFunction] = None,
558 response_deserializer: Optional[DeserializingFunction] = None,
559 _registered_method: Optional[bool] = False,
560 ) -> StreamStreamMultiCallable:
561 return StreamStreamMultiCallable(
562 self._channel,
563 _common.encode(method),
564 request_serializer,
565 response_deserializer,
566 self._stream_stream_interceptors,
567 [self],
568 self._loop,
569 )
572def insecure_channel(
573 target: str,
574 options: Optional[ChannelArgumentType] = None,
575 compression: Optional[grpc.Compression] = None,
576 interceptors: Optional[Sequence[ClientInterceptor]] = None,
577):
578 """Creates an insecure asynchronous Channel to a server.
580 Args:
581 target: The server address
582 options: An optional list of key-value pairs (:term:`channel_arguments`
583 in gRPC Core runtime) to configure the channel.
584 compression: An optional value indicating the compression method to be
585 used over the lifetime of the channel.
586 interceptors: An optional sequence of interceptors that will be executed for
587 any call executed with this channel.
589 Returns:
590 A Channel.
591 """
592 return Channel(
593 target,
594 () if options is None else options,
595 None,
596 compression,
597 interceptors,
598 )
601def secure_channel(
602 target: str,
603 credentials: grpc.ChannelCredentials,
604 options: Optional[ChannelArgumentType] = None,
605 compression: Optional[grpc.Compression] = None,
606 interceptors: Optional[Sequence[ClientInterceptor]] = None,
607):
608 """Creates a secure asynchronous Channel to a server.
610 Args:
611 target: The server address.
612 credentials: A ChannelCredentials instance.
613 options: An optional list of key-value pairs (:term:`channel_arguments`
614 in gRPC Core runtime) to configure the channel.
615 compression: An optional value indicating the compression method to be
616 used over the lifetime of the channel.
617 interceptors: An optional sequence of interceptors that will be executed for
618 any call executed with this channel.
620 Returns:
621 An aio.Channel.
622 """
623 return Channel(
624 target,
625 () if options is None else options,
626 credentials._credentials,
627 compression,
628 interceptors,
629 )