[Type hints] Fix type hints in aio._interceptors.py (#28770)

As per the title.

<!--

Your pull request will be routed to the following person by default for
triaging.
If you know who should review your pull request, please remove the
mentioning below.

-->

@donnadionne
pull/34040/head
Yusuke Nishioka 1 year ago committed by GitHub
parent 06d0a6c0fd
commit 4a24d8250f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 10
      src/python/grpcio/grpc/aio/_interceptor.py

@ -741,7 +741,7 @@ class InterceptedUnaryStreamCall(
# pylint: disable=too-many-arguments
async def _invoke(
self,
interceptors: Sequence[UnaryUnaryClientInterceptor],
interceptors: Sequence[UnaryStreamClientInterceptor],
method: bytes,
timeout: Optional[float],
metadata: Optional[Metadata],
@ -757,7 +757,7 @@ class InterceptedUnaryStreamCall(
interceptors: List[UnaryStreamClientInterceptor],
client_call_details: ClientCallDetails,
request: RequestType,
) -> _base_call.UnaryUnaryCall:
) -> _base_call.UnaryStreamCall:
if interceptors:
continuation = functools.partial(
_run_interceptor, interceptors[1:]
@ -874,7 +874,7 @@ class InterceptedStreamUnaryCall(
"""Run the RPC call wrapped in interceptors"""
async def _run_interceptor(
interceptors: Iterator[UnaryUnaryClientInterceptor],
interceptors: Iterator[StreamUnaryClientInterceptor],
client_call_details: ClientCallDetails,
request_iterator: RequestIterableType,
) -> _base_call.StreamUnaryCall:
@ -921,7 +921,9 @@ class InterceptedStreamStreamCall(
_loop: asyncio.AbstractEventLoop
_channel: cygrpc.AioChannel
_last_returned_call_from_interceptors = Optional[_base_call.UnaryStreamCall]
_last_returned_call_from_interceptors = Optional[
_base_call.StreamStreamCall
]
# pylint: disable=too-many-arguments
def __init__(

Loading…
Cancel
Save