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"""Interceptors implementation of gRPC Asyncio Python."""
16from __future__ import annotations
18from abc import ABCMeta
19from abc import abstractmethod
20import asyncio
21import collections
22import functools
23from typing import (
24 AsyncIterable,
25 AsyncIterator,
26 Awaitable,
27 Callable,
28 List,
29 Optional,
30 Sequence,
31 Union,
32)
34import grpc
35from grpc._cython import cygrpc
37from . import _base_call
38from ._call import AioRpcError
39from ._call import StreamStreamCall
40from ._call import StreamUnaryCall
41from ._call import UnaryStreamCall
42from ._call import UnaryUnaryCall
43from ._call import _API_STYLE_ERROR
44from ._call import _RPC_ALREADY_FINISHED_DETAILS
45from ._call import _RPC_HALF_CLOSED_DETAILS
46from ._metadata import Metadata
47from ._typing import DeserializingFunction
48from ._typing import DoneCallbackType
49from ._typing import EOFType
50from ._typing import RequestIterableType
51from ._typing import RequestType
52from ._typing import ResponseIterableType
53from ._typing import ResponseType
54from ._typing import SerializingFunction
55from ._utils import _timeout_to_deadline
57_LOCAL_CANCELLATION_DETAILS = "Locally cancelled by application!"
60class ServerInterceptor(metaclass=ABCMeta):
61 """Affords intercepting incoming RPCs on the service-side.
63 This is an EXPERIMENTAL API.
64 """
66 @abstractmethod
67 async def intercept_service(
68 self,
69 continuation: Callable[
70 [grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler]
71 ],
72 handler_call_details: grpc.HandlerCallDetails,
73 ) -> grpc.RpcMethodHandler:
74 """Intercepts incoming RPCs before handing them over to a handler.
76 State can be passed from an interceptor to downstream interceptors
77 via contextvars. The first interceptor is called from an empty
78 contextvars.Context, and the same Context is used for downstream
79 interceptors and for the final handler call. Note that there are no
80 guarantees that interceptors and handlers will be called from the
81 same thread.
83 Args:
84 continuation: A function that takes a HandlerCallDetails and
85 proceeds to invoke the next interceptor in the chain, if any,
86 or the RPC handler lookup logic, with the call details passed
87 as an argument, and returns an RpcMethodHandler instance if
88 the RPC is considered serviced, or None otherwise.
89 handler_call_details: A HandlerCallDetails describing the RPC.
91 Returns:
92 An RpcMethodHandler with which the RPC may be serviced if the
93 interceptor chooses to service this RPC, or None otherwise.
94 """
97class ClientCallDetails(
98 collections.namedtuple(
99 "ClientCallDetails",
100 ("method", "timeout", "metadata", "credentials", "wait_for_ready"),
101 ),
102 grpc.ClientCallDetails,
103):
104 """Describes an RPC to be invoked.
106 This is an EXPERIMENTAL API.
108 Args:
109 method: The method name of the RPC.
110 timeout: An optional duration of time in seconds to allow for the RPC.
111 metadata: Optional metadata to be transmitted to the service-side of
112 the RPC.
113 credentials: An optional CallCredentials for the RPC.
114 wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism.
115 """
117 method: bytes
118 timeout: Optional[float]
119 metadata: Optional[Metadata]
120 credentials: Optional[grpc.CallCredentials]
121 wait_for_ready: Optional[bool]
124class ClientInterceptor(metaclass=ABCMeta):
125 """Base class used for all Aio Client Interceptor classes"""
128class UnaryUnaryClientInterceptor(ClientInterceptor, metaclass=ABCMeta):
129 """Affords intercepting unary-unary invocations."""
131 @abstractmethod
132 async def intercept_unary_unary(
133 self,
134 continuation: Callable[
135 [ClientCallDetails, RequestType], UnaryUnaryCall
136 ],
137 client_call_details: ClientCallDetails,
138 request: RequestType,
139 ) -> Union[UnaryUnaryCall, ResponseType]:
140 """Intercepts a unary-unary invocation asynchronously.
142 Args:
143 continuation: A coroutine that proceeds with the invocation by
144 executing the next interceptor in the chain or invoking the
145 actual RPC on the underlying Channel. It is the interceptor's
146 responsibility to call it if it decides to move the RPC forward.
147 The interceptor can use
148 `call = await continuation(client_call_details, request)`
149 to continue with the RPC. `continuation` returns the call to the
150 RPC.
151 client_call_details: A ClientCallDetails object describing the
152 outgoing RPC.
153 request: The request value for the RPC.
155 Returns:
156 An object with the RPC response.
158 Raises:
159 AioRpcError: Indicating that the RPC terminated with non-OK status.
160 asyncio.CancelledError: Indicating that the RPC was canceled.
161 """
164class UnaryStreamClientInterceptor(ClientInterceptor, metaclass=ABCMeta):
165 """Affords intercepting unary-stream invocations."""
167 @abstractmethod
168 async def intercept_unary_stream(
169 self,
170 continuation: Callable[
171 [ClientCallDetails, RequestType], UnaryStreamCall
172 ],
173 client_call_details: ClientCallDetails,
174 request: RequestType,
175 ) -> Union[ResponseIterableType, UnaryStreamCall]:
176 """Intercepts a unary-stream invocation asynchronously.
178 The function could return the call object or an asynchronous
179 iterator, in case of being an asyncrhonous iterator this will
180 become the source of the reads done by the caller.
182 Args:
183 continuation: A coroutine that proceeds with the invocation by
184 executing the next interceptor in the chain or invoking the
185 actual RPC on the underlying Channel. It is the interceptor's
186 responsibility to call it if it decides to move the RPC forward.
187 The interceptor can use
188 `call = await continuation(client_call_details, request)`
189 to continue with the RPC. `continuation` returns the call to the
190 RPC.
191 client_call_details: A ClientCallDetails object describing the
192 outgoing RPC.
193 request: The request value for the RPC.
195 Returns:
196 The RPC Call or an asynchronous iterator.
198 Raises:
199 AioRpcError: Indicating that the RPC terminated with non-OK status.
200 asyncio.CancelledError: Indicating that the RPC was canceled.
201 """
204class StreamUnaryClientInterceptor(ClientInterceptor, metaclass=ABCMeta):
205 """Affords intercepting stream-unary invocations."""
207 @abstractmethod
208 async def intercept_stream_unary(
209 self,
210 continuation: Callable[
211 [ClientCallDetails, RequestType], StreamUnaryCall
212 ],
213 client_call_details: ClientCallDetails,
214 request_iterator: RequestIterableType,
215 ) -> StreamUnaryCall:
216 """Intercepts a stream-unary invocation asynchronously.
218 Within the interceptor the usage of the call methods like `write` or
219 even awaiting the call should be done carefully, since the caller
220 could be expecting an untouched call, for example for start writing
221 messages to it.
223 Args:
224 continuation: A coroutine that proceeds with the invocation by
225 executing the next interceptor in the chain or invoking the
226 actual RPC on the underlying Channel. It is the interceptor's
227 responsibility to call it if it decides to move the RPC forward.
228 The interceptor can use
229 `call = await continuation(client_call_details, request_iterator)`
230 to continue with the RPC. `continuation` returns the call to the
231 RPC.
232 client_call_details: A ClientCallDetails object describing the
233 outgoing RPC.
234 request_iterator: The request iterator that will produce requests
235 for the RPC.
237 Returns:
238 The RPC Call.
240 Raises:
241 AioRpcError: Indicating that the RPC terminated with non-OK status.
242 asyncio.CancelledError: Indicating that the RPC was canceled.
243 """
246class StreamStreamClientInterceptor(ClientInterceptor, metaclass=ABCMeta):
247 """Affords intercepting stream-stream invocations."""
249 @abstractmethod
250 async def intercept_stream_stream(
251 self,
252 continuation: Callable[
253 [ClientCallDetails, RequestType], StreamStreamCall
254 ],
255 client_call_details: ClientCallDetails,
256 request_iterator: RequestIterableType,
257 ) -> Union[ResponseIterableType, StreamStreamCall]:
258 """Intercepts a stream-stream invocation asynchronously.
260 Within the interceptor the usage of the call methods like `write` or
261 even awaiting the call should be done carefully, since the caller
262 could be expecting an untouched call, for example for start writing
263 messages to it.
265 The function could return the call object or an asynchronous
266 iterator, in case of being an asyncrhonous iterator this will
267 become the source of the reads done by the caller.
269 Args:
270 continuation: A coroutine that proceeds with the invocation by
271 executing the next interceptor in the chain or invoking the
272 actual RPC on the underlying Channel. It is the interceptor's
273 responsibility to call it if it decides to move the RPC forward.
274 The interceptor can use
275 `call = await continuation(client_call_details, request_iterator)`
276 to continue with the RPC. `continuation` returns the call to the
277 RPC.
278 client_call_details: A ClientCallDetails object describing the
279 outgoing RPC.
280 request_iterator: The request iterator that will produce requests
281 for the RPC.
283 Returns:
284 The RPC Call or an asynchronous iterator.
286 Raises:
287 AioRpcError: Indicating that the RPC terminated with non-OK status.
288 asyncio.CancelledError: Indicating that the RPC was canceled.
289 """
292class InterceptedCall:
293 """Base implementation for all intercepted call arities.
295 Interceptors might have some work to do before the RPC invocation with
296 the capacity of changing the invocation parameters, and some work to do
297 after the RPC invocation with the capacity for accessing to the wrapped
298 `UnaryUnaryCall`.
300 It handles also early and later cancellations, when the RPC has not even
301 started and the execution is still held by the interceptors or when the
302 RPC has finished but again the execution is still held by the interceptors.
304 Once the RPC is finally executed, all methods are finally done against the
305 intercepted call, being at the same time the same call returned to the
306 interceptors.
308 As a base class for all of the interceptors implements the logic around
309 final status, metadata and cancellation.
310 """
312 _interceptors_task: asyncio.Task
313 _pending_add_done_callbacks: Sequence[DoneCallbackType]
315 def __init__(self, interceptors_task: asyncio.Task) -> None:
316 self._interceptors_task = interceptors_task
317 self._pending_add_done_callbacks = []
318 self._interceptors_task.add_done_callback(
319 self._fire_or_add_pending_done_callbacks
320 )
322 def __del__(self):
323 self.cancel()
325 def _fire_or_add_pending_done_callbacks(
326 self, interceptors_task: asyncio.Task
327 ) -> None:
328 if not self._pending_add_done_callbacks:
329 return
331 call_completed = False
333 if interceptors_task.cancelled() or (
334 interceptors_task.done()
335 and interceptors_task.exception() is not None
336 ):
337 call_completed = True
338 else:
339 call = interceptors_task.result()
340 if call.done():
341 call_completed = True
343 if call_completed:
344 for callback in self._pending_add_done_callbacks:
345 callback(self)
346 else:
347 for callback in self._pending_add_done_callbacks:
348 callback = functools.partial(
349 self._wrap_add_done_callback, callback
350 )
351 call.add_done_callback(callback)
353 self._pending_add_done_callbacks = []
355 def _wrap_add_done_callback(
356 self, callback: DoneCallbackType, unused_call: _base_call.Call
357 ) -> None:
358 callback(self)
360 def cancel(self) -> bool:
361 if self._interceptors_task.cancelled():
362 return False
364 if not self._interceptors_task.done():
365 # There is no yet the intercepted call available,
366 # Trying to cancel it by using the generic Asyncio
367 # cancellation method.
368 return self._interceptors_task.cancel()
370 if self._interceptors_task.exception() is not None:
371 return False
373 call = self._interceptors_task.result()
375 return call.cancel()
377 def cancelled(self) -> bool:
378 if self._interceptors_task.cancelled():
379 return True
380 if not self._interceptors_task.done():
381 return False
383 exc = self._interceptors_task.exception()
384 if exc is not None:
385 if isinstance(exc, AioRpcError):
386 return exc.code() == grpc.StatusCode.CANCELLED
387 return False
389 call = self._interceptors_task.result()
391 return call.cancelled()
393 def done(self) -> bool:
394 if self._interceptors_task.cancelled():
395 return True
396 if not self._interceptors_task.done():
397 return False
398 if self._interceptors_task.exception() is not None:
399 return True
401 call = self._interceptors_task.result()
403 return call.done()
405 def add_done_callback(self, callback: DoneCallbackType) -> None:
406 if not self._interceptors_task.done():
407 self._pending_add_done_callbacks.append(callback)
408 return
410 if (
411 self._interceptors_task.cancelled()
412 or self._interceptors_task.exception() is not None
413 ):
414 callback(self)
415 return
417 call = self._interceptors_task.result()
419 if call.done():
420 callback(self)
421 else:
422 callback = functools.partial(self._wrap_add_done_callback, callback)
423 call.add_done_callback(callback)
425 def time_remaining(self) -> Optional[float]:
426 raise NotImplementedError()
428 async def initial_metadata(self) -> Optional[Metadata]:
429 try:
430 call = await self._interceptors_task
431 except AioRpcError as err:
432 return err.initial_metadata()
433 except asyncio.CancelledError:
434 return None
436 return await call.initial_metadata()
438 async def trailing_metadata(self) -> Optional[Metadata]:
439 try:
440 call = await self._interceptors_task
441 except AioRpcError as err:
442 return err.trailing_metadata()
443 except asyncio.CancelledError:
444 return None
446 return await call.trailing_metadata()
448 async def code(self) -> grpc.StatusCode:
449 try:
450 call = await self._interceptors_task
451 except AioRpcError as err:
452 return err.code()
453 except asyncio.CancelledError:
454 return grpc.StatusCode.CANCELLED
456 return await call.code()
458 async def details(self) -> str:
459 try:
460 call = await self._interceptors_task
461 except AioRpcError as err:
462 return err.details()
463 except asyncio.CancelledError:
464 return _LOCAL_CANCELLATION_DETAILS
466 return await call.details()
468 async def debug_error_string(self) -> Optional[str]:
469 try:
470 call = await self._interceptors_task
471 except AioRpcError as err:
472 return err.debug_error_string()
473 except asyncio.CancelledError:
474 return ""
476 return await call.debug_error_string()
478 async def wait_for_connection(self) -> None:
479 call = await self._interceptors_task
480 return await call.wait_for_connection()
483class _InterceptedUnaryResponseMixin:
484 def __await__(self):
485 call = yield from self._interceptors_task.__await__()
486 response = yield from call.__await__()
487 return response
490class _InterceptedStreamResponseMixin:
491 _response_aiter: Optional[AsyncIterable[ResponseType]]
493 def _init_stream_response_mixin(self) -> None:
494 # Is initialized later, otherwise if the iterator is not finally
495 # consumed a logging warning is emitted by Asyncio.
496 self._response_aiter = None
498 async def _wait_for_interceptor_task_response_iterator(
499 self,
500 ) -> ResponseType:
501 call = await self._interceptors_task
502 async for response in call:
503 yield response
505 def __aiter__(self) -> AsyncIterator[ResponseType]:
506 if self._response_aiter is None:
507 self._response_aiter = (
508 self._wait_for_interceptor_task_response_iterator()
509 )
510 return self._response_aiter
512 async def read(self) -> Union[EOFType, ResponseType]:
513 if self._response_aiter is None:
514 self._response_aiter = (
515 self._wait_for_interceptor_task_response_iterator()
516 )
517 try:
518 return await self._response_aiter.asend(None)
519 except StopAsyncIteration:
520 return cygrpc.EOF
523class _InterceptedStreamRequestMixin:
524 _write_to_iterator_async_gen: Optional[AsyncIterable[RequestType]]
525 _write_to_iterator_queue: Optional[asyncio.Queue]
526 _status_code_task: Optional[asyncio.Task]
528 _FINISH_ITERATOR_SENTINEL = object()
530 def _init_stream_request_mixin(
531 self, request_iterator: Optional[RequestIterableType]
532 ) -> RequestIterableType:
533 if request_iterator is None:
534 # We provide our own request iterator which is a proxy
535 # of the futures writes that will be done by the caller.
536 self._write_to_iterator_queue = asyncio.Queue(maxsize=1)
537 self._write_to_iterator_async_gen = (
538 self._proxy_writes_as_request_iterator()
539 )
540 self._status_code_task = None
541 request_iterator = self._write_to_iterator_async_gen
542 else:
543 self._write_to_iterator_queue = None
545 return request_iterator
547 async def _proxy_writes_as_request_iterator(self):
548 await self._interceptors_task
550 while True:
551 value = await self._write_to_iterator_queue.get()
552 if (
553 value
554 is _InterceptedStreamRequestMixin._FINISH_ITERATOR_SENTINEL
555 ):
556 break
557 yield value
559 async def _write_to_iterator_queue_interruptible(
560 self,
561 request: RequestType,
562 call: _base_call.Call,
563 ):
564 # Write the specified 'request' to the request iterator queue using the
565 # specified 'call' to allow for interruption of the write in the case
566 # of abrupt termination of the call.
567 if self._status_code_task is None:
568 self._status_code_task = self._loop.create_task(call.code())
570 put_task = self._loop.create_task(
571 self._write_to_iterator_queue.put(request)
572 )
573 try:
574 await asyncio.wait(
575 (
576 put_task,
577 self._status_code_task,
578 ),
579 return_when=asyncio.FIRST_COMPLETED,
580 )
581 finally:
582 # If put_task is not done, it means the RPC has completed.
583 # This means the RPC terminated before the message could
584 # be queued. We must cancel put_task to prevent a leak.
585 if not put_task.done():
586 put_task.cancel()
588 async def write(self, request: RequestType) -> None:
589 # If no queue was created it means that requests
590 # should be expected through an iterators provided
591 # by the caller.
592 if self._write_to_iterator_queue is None:
593 raise cygrpc.UsageError(_API_STYLE_ERROR)
595 try:
596 call = await self._interceptors_task
597 except (asyncio.CancelledError, AioRpcError):
598 raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS)
600 if call.done():
601 raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS)
602 if call._done_writing_flag:
603 raise asyncio.InvalidStateError(_RPC_HALF_CLOSED_DETAILS)
605 await self._write_to_iterator_queue_interruptible(request, call)
607 if call.done():
608 raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS)
610 async def done_writing(self) -> None:
611 """Signal peer that client is done writing.
613 This method is idempotent.
614 """
615 # If no queue was created it means that requests
616 # should be expected through an iterators provided
617 # by the caller.
618 if self._write_to_iterator_queue is None:
619 raise cygrpc.UsageError(_API_STYLE_ERROR)
621 try:
622 call = await self._interceptors_task
623 except asyncio.CancelledError:
624 raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS)
626 await self._write_to_iterator_queue_interruptible(
627 _InterceptedStreamRequestMixin._FINISH_ITERATOR_SENTINEL, call
628 )
631def _resolve_registered_call_handle(
632 channel: cygrpc.AioChannel,
633 original_method: bytes,
634 new_method: bytes,
635 registered_call_handle: int,
636) -> int:
637 """Returns the registered call handle matching the outgoing method."""
638 resolved_registered_call_handle = registered_call_handle
639 if new_method != original_method and registered_call_handle != 0:
640 resolved_registered_call_handle = channel.get_registered_call_handle(
641 new_method,
642 )
643 return resolved_registered_call_handle
646class InterceptedUnaryUnaryCall(
647 _InterceptedUnaryResponseMixin, InterceptedCall, _base_call.UnaryUnaryCall
648):
649 """Used for running a `UnaryUnaryCall` wrapped by interceptors.
651 For the `__await__` method is it is proxied to the intercepted call only when
652 the interceptor task is finished.
653 """
655 _loop: asyncio.AbstractEventLoop
656 _channel: cygrpc.AioChannel
658 def __init__(
659 self,
660 interceptors: Sequence[UnaryUnaryClientInterceptor],
661 request: RequestType,
662 timeout: Optional[float],
663 metadata: Metadata,
664 credentials: Optional[grpc.CallCredentials],
665 wait_for_ready: Optional[bool],
666 channel: cygrpc.AioChannel,
667 method: bytes,
668 request_serializer: Optional[SerializingFunction],
669 response_deserializer: Optional[DeserializingFunction],
670 loop: asyncio.AbstractEventLoop,
671 registered_call_handle: int = 0,
672 ) -> None:
673 self._loop = loop
674 self._channel = channel
675 self._registered_call_handle = registered_call_handle
676 interceptors_task = loop.create_task(
677 self._invoke(
678 interceptors,
679 method,
680 timeout,
681 metadata,
682 credentials,
683 wait_for_ready,
684 request,
685 request_serializer,
686 response_deserializer,
687 )
688 )
689 super().__init__(interceptors_task)
691 async def _invoke(
692 self,
693 interceptors: Sequence[UnaryUnaryClientInterceptor],
694 method: bytes,
695 timeout: Optional[float],
696 metadata: Optional[Metadata],
697 credentials: Optional[grpc.CallCredentials],
698 wait_for_ready: Optional[bool],
699 request: RequestType,
700 request_serializer: Optional[SerializingFunction],
701 response_deserializer: Optional[DeserializingFunction],
702 ) -> Union[UnaryUnaryCall, UnaryUnaryCallResponse]:
703 """Run the RPC call wrapped in interceptors"""
704 client_call_details = ClientCallDetails(
705 method, timeout, metadata, credentials, wait_for_ready
706 )
707 return await self._run_interceptor(
708 list(interceptors),
709 method,
710 request_serializer,
711 response_deserializer,
712 client_call_details,
713 request,
714 )
716 async def _run_interceptor(
717 self,
718 interceptors: List[UnaryUnaryClientInterceptor],
719 method: bytes,
720 request_serializer: Optional[SerializingFunction],
721 response_deserializer: Optional[DeserializingFunction],
722 client_call_details: ClientCallDetails,
723 request: RequestType,
724 ) -> Union[UnaryUnaryCall, UnaryUnaryCallResponse]:
725 if interceptors:
726 continuation = functools.partial(
727 self._run_interceptor,
728 interceptors[1:],
729 method,
730 request_serializer,
731 response_deserializer,
732 )
733 call_or_response = await interceptors[0].intercept_unary_unary(
734 continuation, client_call_details, request
735 )
737 if isinstance(call_or_response, _base_call.UnaryUnaryCall):
738 return call_or_response
739 return UnaryUnaryCallResponse(call_or_response)
741 registered_call_handle = _resolve_registered_call_handle(
742 self._channel,
743 method,
744 client_call_details.method,
745 self._registered_call_handle,
746 )
748 return UnaryUnaryCall(
749 request,
750 _timeout_to_deadline(client_call_details.timeout),
751 client_call_details.metadata,
752 client_call_details.credentials,
753 client_call_details.wait_for_ready,
754 self._channel,
755 client_call_details.method,
756 request_serializer,
757 response_deserializer,
758 self._loop,
759 registered_call_handle,
760 )
762 def time_remaining(self) -> Optional[float]:
763 raise NotImplementedError()
766class InterceptedUnaryStreamCall(
767 _InterceptedStreamResponseMixin, InterceptedCall, _base_call.UnaryStreamCall
768):
769 """Used for running a `UnaryStreamCall` wrapped by interceptors."""
771 _loop: asyncio.AbstractEventLoop
772 _channel: cygrpc.AioChannel
773 _last_returned_call_from_interceptors = Optional[_base_call.UnaryStreamCall]
775 def __init__(
776 self,
777 interceptors: Sequence[UnaryStreamClientInterceptor],
778 request: RequestType,
779 timeout: Optional[float],
780 metadata: Metadata,
781 credentials: Optional[grpc.CallCredentials],
782 wait_for_ready: Optional[bool],
783 channel: cygrpc.AioChannel,
784 method: bytes,
785 request_serializer: Optional[SerializingFunction],
786 response_deserializer: Optional[DeserializingFunction],
787 loop: asyncio.AbstractEventLoop,
788 registered_call_handle: int = 0,
789 ) -> None:
790 self._loop = loop
791 self._channel = channel
792 self._registered_call_handle = registered_call_handle
793 self._init_stream_response_mixin()
794 self._last_returned_call_from_interceptors = None
795 interceptors_task = loop.create_task(
796 self._invoke(
797 interceptors,
798 method,
799 timeout,
800 metadata,
801 credentials,
802 wait_for_ready,
803 request,
804 request_serializer,
805 response_deserializer,
806 )
807 )
808 super().__init__(interceptors_task)
810 async def _invoke(
811 self,
812 interceptors: Sequence[UnaryStreamClientInterceptor],
813 method: bytes,
814 timeout: Optional[float],
815 metadata: Optional[Metadata],
816 credentials: Optional[grpc.CallCredentials],
817 wait_for_ready: Optional[bool],
818 request: RequestType,
819 request_serializer: Optional[SerializingFunction],
820 response_deserializer: Optional[DeserializingFunction],
821 ) -> Union[UnaryStreamCall, UnaryStreamCallResponseIterator]:
822 """Run the RPC call wrapped in interceptors"""
823 client_call_details = ClientCallDetails(
824 method, timeout, metadata, credentials, wait_for_ready
825 )
826 return await self._run_interceptor(
827 list(interceptors),
828 method,
829 request_serializer,
830 response_deserializer,
831 client_call_details,
832 request,
833 )
835 async def _run_interceptor(
836 self,
837 interceptors: List[UnaryStreamClientInterceptor],
838 method: bytes,
839 request_serializer: Optional[SerializingFunction],
840 response_deserializer: Optional[DeserializingFunction],
841 client_call_details: ClientCallDetails,
842 request: RequestType,
843 ) -> Union[UnaryStreamCall, UnaryStreamCallResponseIterator]:
844 if interceptors:
845 continuation = functools.partial(
846 self._run_interceptor,
847 interceptors[1:],
848 method,
849 request_serializer,
850 response_deserializer,
851 )
853 call_or_response_iterator = await interceptors[
854 0
855 ].intercept_unary_stream(continuation, client_call_details, request)
857 if isinstance(
858 call_or_response_iterator, _base_call.UnaryStreamCall
859 ):
860 self._last_returned_call_from_interceptors = (
861 call_or_response_iterator
862 )
863 else:
864 self._last_returned_call_from_interceptors = (
865 UnaryStreamCallResponseIterator(
866 self._last_returned_call_from_interceptors,
867 call_or_response_iterator,
868 )
869 )
870 return self._last_returned_call_from_interceptors
872 registered_call_handle = _resolve_registered_call_handle(
873 self._channel,
874 method,
875 client_call_details.method,
876 self._registered_call_handle,
877 )
879 self._last_returned_call_from_interceptors = UnaryStreamCall(
880 request,
881 _timeout_to_deadline(client_call_details.timeout),
882 client_call_details.metadata,
883 client_call_details.credentials,
884 client_call_details.wait_for_ready,
885 self._channel,
886 client_call_details.method,
887 request_serializer,
888 response_deserializer,
889 self._loop,
890 registered_call_handle,
891 )
893 return self._last_returned_call_from_interceptors
895 def time_remaining(self) -> Optional[float]:
896 raise NotImplementedError()
899class InterceptedStreamUnaryCall(
900 _InterceptedUnaryResponseMixin,
901 _InterceptedStreamRequestMixin,
902 InterceptedCall,
903 _base_call.StreamUnaryCall,
904):
905 """Used for running a `StreamUnaryCall` wrapped by interceptors.
907 For the `__await__` method is it is proxied to the intercepted call only when
908 the interceptor task is finished.
909 """
911 _loop: asyncio.AbstractEventLoop
912 _channel: cygrpc.AioChannel
914 def __init__(
915 self,
916 interceptors: Sequence[StreamUnaryClientInterceptor],
917 request_iterator: Optional[RequestIterableType],
918 timeout: Optional[float],
919 metadata: Metadata,
920 credentials: Optional[grpc.CallCredentials],
921 wait_for_ready: Optional[bool],
922 channel: cygrpc.AioChannel,
923 method: bytes,
924 request_serializer: Optional[SerializingFunction],
925 response_deserializer: Optional[DeserializingFunction],
926 loop: asyncio.AbstractEventLoop,
927 registered_call_handle: int = 0,
928 ) -> None:
929 self._loop = loop
930 self._channel = channel
931 self._registered_call_handle = registered_call_handle
932 request_iterator = self._init_stream_request_mixin(request_iterator)
933 interceptors_task = loop.create_task(
934 self._invoke(
935 interceptors,
936 method,
937 timeout,
938 metadata,
939 credentials,
940 wait_for_ready,
941 request_iterator,
942 request_serializer,
943 response_deserializer,
944 )
945 )
946 super().__init__(interceptors_task)
948 async def _invoke(
949 self,
950 interceptors: Sequence[StreamUnaryClientInterceptor],
951 method: bytes,
952 timeout: Optional[float],
953 metadata: Optional[Metadata],
954 credentials: Optional[grpc.CallCredentials],
955 wait_for_ready: Optional[bool],
956 request_iterator: RequestIterableType,
957 request_serializer: Optional[SerializingFunction],
958 response_deserializer: Optional[DeserializingFunction],
959 ) -> StreamUnaryCall:
960 """Run the RPC call wrapped in interceptors"""
961 client_call_details = ClientCallDetails(
962 method, timeout, metadata, credentials, wait_for_ready
963 )
964 return await self._run_interceptor(
965 list(interceptors),
966 method,
967 request_serializer,
968 response_deserializer,
969 client_call_details,
970 request_iterator,
971 )
973 async def _run_interceptor(
974 self,
975 interceptors: Sequence[StreamUnaryClientInterceptor],
976 method: bytes,
977 request_serializer: Optional[SerializingFunction],
978 response_deserializer: Optional[DeserializingFunction],
979 client_call_details: ClientCallDetails,
980 request_iterator: RequestIterableType,
981 ) -> _base_call.StreamUnaryCall:
982 if interceptors:
983 continuation = functools.partial(
984 self._run_interceptor,
985 interceptors[1:],
986 method,
987 request_serializer,
988 response_deserializer,
989 )
991 return await interceptors[0].intercept_stream_unary(
992 continuation, client_call_details, request_iterator
993 )
995 registered_call_handle = _resolve_registered_call_handle(
996 self._channel,
997 method,
998 client_call_details.method,
999 self._registered_call_handle,
1000 )
1002 return StreamUnaryCall(
1003 request_iterator,
1004 _timeout_to_deadline(client_call_details.timeout),
1005 client_call_details.metadata,
1006 client_call_details.credentials,
1007 client_call_details.wait_for_ready,
1008 self._channel,
1009 client_call_details.method,
1010 request_serializer,
1011 response_deserializer,
1012 self._loop,
1013 registered_call_handle,
1014 )
1016 def time_remaining(self) -> Optional[float]:
1017 raise NotImplementedError()
1020class InterceptedStreamStreamCall(
1021 _InterceptedStreamResponseMixin,
1022 _InterceptedStreamRequestMixin,
1023 InterceptedCall,
1024 _base_call.StreamStreamCall,
1025):
1026 """Used for running a `StreamStreamCall` wrapped by interceptors."""
1028 _loop: asyncio.AbstractEventLoop
1029 _channel: cygrpc.AioChannel
1030 _last_returned_call_from_interceptors = Optional[
1031 _base_call.StreamStreamCall
1032 ]
1034 def __init__(
1035 self,
1036 interceptors: Sequence[StreamStreamClientInterceptor],
1037 request_iterator: Optional[RequestIterableType],
1038 timeout: Optional[float],
1039 metadata: Metadata,
1040 credentials: Optional[grpc.CallCredentials],
1041 wait_for_ready: Optional[bool],
1042 channel: cygrpc.AioChannel,
1043 method: bytes,
1044 request_serializer: Optional[SerializingFunction],
1045 response_deserializer: Optional[DeserializingFunction],
1046 loop: asyncio.AbstractEventLoop,
1047 registered_call_handle: int = 0,
1048 ) -> None:
1049 self._loop = loop
1050 self._channel = channel
1051 self._registered_call_handle = registered_call_handle
1052 self._init_stream_response_mixin()
1053 request_iterator = self._init_stream_request_mixin(request_iterator)
1054 self._last_returned_call_from_interceptors = None
1055 interceptors_task = loop.create_task(
1056 self._invoke(
1057 interceptors,
1058 method,
1059 timeout,
1060 metadata,
1061 credentials,
1062 wait_for_ready,
1063 request_iterator,
1064 request_serializer,
1065 response_deserializer,
1066 )
1067 )
1068 super().__init__(interceptors_task)
1070 async def _invoke(
1071 self,
1072 interceptors: Sequence[StreamStreamClientInterceptor],
1073 method: bytes,
1074 timeout: Optional[float],
1075 metadata: Optional[Metadata],
1076 credentials: Optional[grpc.CallCredentials],
1077 wait_for_ready: Optional[bool],
1078 request_iterator: RequestIterableType,
1079 request_serializer: Optional[SerializingFunction],
1080 response_deserializer: Optional[DeserializingFunction],
1081 ) -> Union[StreamStreamCall, StreamStreamCallResponseIterator]:
1082 """Run the RPC call wrapped in interceptors"""
1083 client_call_details = ClientCallDetails(
1084 method, timeout, metadata, credentials, wait_for_ready
1085 )
1086 return await self._run_interceptor(
1087 list(interceptors),
1088 method,
1089 request_serializer,
1090 response_deserializer,
1091 client_call_details,
1092 request_iterator,
1093 )
1095 async def _run_interceptor(
1096 self,
1097 interceptors: List[StreamStreamClientInterceptor],
1098 method: bytes,
1099 request_serializer: Optional[SerializingFunction],
1100 response_deserializer: Optional[DeserializingFunction],
1101 client_call_details: ClientCallDetails,
1102 request_iterator: RequestIterableType,
1103 ) -> Union[StreamStreamCall, StreamStreamCallResponseIterator]:
1104 if interceptors:
1105 continuation = functools.partial(
1106 self._run_interceptor,
1107 interceptors[1:],
1108 method,
1109 request_serializer,
1110 response_deserializer,
1111 )
1113 call_or_response_iterator = await interceptors[
1114 0
1115 ].intercept_stream_stream(
1116 continuation, client_call_details, request_iterator
1117 )
1119 if isinstance(
1120 call_or_response_iterator, _base_call.StreamStreamCall
1121 ):
1122 self._last_returned_call_from_interceptors = (
1123 call_or_response_iterator
1124 )
1125 else:
1126 self._last_returned_call_from_interceptors = (
1127 StreamStreamCallResponseIterator(
1128 self._last_returned_call_from_interceptors,
1129 call_or_response_iterator,
1130 )
1131 )
1132 return self._last_returned_call_from_interceptors
1134 registered_call_handle = _resolve_registered_call_handle(
1135 self._channel,
1136 method,
1137 client_call_details.method,
1138 self._registered_call_handle,
1139 )
1141 self._last_returned_call_from_interceptors = StreamStreamCall(
1142 request_iterator,
1143 _timeout_to_deadline(client_call_details.timeout),
1144 client_call_details.metadata,
1145 client_call_details.credentials,
1146 client_call_details.wait_for_ready,
1147 self._channel,
1148 client_call_details.method,
1149 request_serializer,
1150 response_deserializer,
1151 self._loop,
1152 registered_call_handle,
1153 )
1154 return self._last_returned_call_from_interceptors
1156 def time_remaining(self) -> Optional[float]:
1157 raise NotImplementedError()
1160class UnaryUnaryCallResponse(_base_call.UnaryUnaryCall):
1161 """Final UnaryUnaryCall class finished with a response."""
1163 _response: ResponseType
1165 def __init__(self, response: ResponseType) -> None:
1166 self._response = response
1168 def cancel(self) -> bool:
1169 return False
1171 def cancelled(self) -> bool:
1172 return False
1174 def done(self) -> bool:
1175 return True
1177 def add_done_callback(self, unused_callback) -> None:
1178 raise NotImplementedError()
1180 def time_remaining(self) -> Optional[float]:
1181 raise NotImplementedError()
1183 async def initial_metadata(self) -> Optional[Metadata]:
1184 return None
1186 async def trailing_metadata(self) -> Optional[Metadata]:
1187 return None
1189 async def code(self) -> grpc.StatusCode:
1190 return grpc.StatusCode.OK
1192 async def details(self) -> str:
1193 return ""
1195 async def debug_error_string(self) -> Optional[str]:
1196 return None
1198 def __await__(self):
1199 if False: # pylint: disable=using-constant-test
1200 # This code path is never used, but a yield statement is needed
1201 # for telling the interpreter that __await__ is a generator.
1202 yield None
1203 return self._response
1205 async def wait_for_connection(self) -> None:
1206 pass
1209class _StreamCallResponseIterator:
1210 _call: Union[_base_call.UnaryStreamCall, _base_call.StreamStreamCall]
1211 _response_iterator: AsyncIterable[ResponseType]
1213 def __init__(
1214 self,
1215 call: Union[_base_call.UnaryStreamCall, _base_call.StreamStreamCall],
1216 response_iterator: AsyncIterable[ResponseType],
1217 ) -> None:
1218 self._response_iterator = response_iterator
1219 self._call = call
1221 def cancel(self) -> bool:
1222 return self._call.cancel()
1224 def cancelled(self) -> bool:
1225 return self._call.cancelled()
1227 def done(self) -> bool:
1228 return self._call.done()
1230 def add_done_callback(self, callback) -> None:
1231 self._call.add_done_callback(callback)
1233 def time_remaining(self) -> Optional[float]:
1234 return self._call.time_remaining()
1236 async def initial_metadata(self) -> Optional[Metadata]:
1237 return await self._call.initial_metadata()
1239 async def trailing_metadata(self) -> Optional[Metadata]:
1240 return await self._call.trailing_metadata()
1242 async def code(self) -> grpc.StatusCode:
1243 return await self._call.code()
1245 async def details(self) -> str:
1246 return await self._call.details()
1248 async def debug_error_string(self) -> Optional[str]:
1249 return await self._call.debug_error_string()
1251 def __aiter__(self):
1252 return self._response_iterator.__aiter__()
1254 async def wait_for_connection(self) -> None:
1255 return await self._call.wait_for_connection()
1258class UnaryStreamCallResponseIterator(
1259 _StreamCallResponseIterator, _base_call.UnaryStreamCall
1260):
1261 """UnaryStreamCall class which uses an alternative response iterator."""
1263 async def read(self) -> Union[EOFType, ResponseType]:
1264 # Behind the scenes everything goes through the
1265 # async iterator. So this path should not be reached.
1266 raise NotImplementedError()
1269class StreamStreamCallResponseIterator(
1270 _StreamCallResponseIterator, _base_call.StreamStreamCall
1271):
1272 """StreamStreamCall class which uses an alternative response iterator."""
1274 async def read(self) -> Union[EOFType, ResponseType]:
1275 # Behind the scenes everything goes through the
1276 # async iterator. So this path should not be reached.
1277 raise NotImplementedError()
1279 async def write(self, request: RequestType) -> None:
1280 # Behind the scenes everything goes through the
1281 # async iterator provided by the InterceptedStreamStreamCall.
1282 # So this path should not be reached.
1283 raise NotImplementedError()
1285 async def done_writing(self) -> None:
1286 # Behind the scenes everything goes through the
1287 # async iterator provided by the InterceptedStreamStreamCall.
1288 # So this path should not be reached.
1289 raise NotImplementedError()
1291 @property
1292 def _done_writing_flag(self) -> bool:
1293 return self._call._done_writing_flag