Skip to content

Commit 0e10493

Browse files
RobinMcCorkellRobinMcCorkell
authored
grpcio: improve client interceptors (#14831)
1 parent af0ee40 commit 0e10493

File tree

2 files changed

+59
-45
lines changed

2 files changed

+59
-45
lines changed

‎stubs/grpcio/grpc/__init__.pyi‎

Lines changed: 23 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -76,13 +76,10 @@ def secure_channel(
7676
) -> Channel: ...
7777

7878
_Interceptor: TypeAlias = (
79-
UnaryUnaryClientInterceptor[_TRequest, _TResponse]
80-
| UnaryStreamClientInterceptor[_TRequest, _TResponse]
81-
| StreamUnaryClientInterceptor[_TRequest, _TResponse]
82-
| StreamStreamClientInterceptor[_TRequest, _TResponse]
79+
UnaryUnaryClientInterceptor | UnaryStreamClientInterceptor | StreamUnaryClientInterceptor | StreamStreamClientInterceptor
8380
)
8481

85-
def intercept_channel(channel: Channel, *interceptors: _Interceptor[_TRequest, _TResponse]) -> Channel: ...
82+
def intercept_channel(channel: Channel, *interceptors: _Interceptor) -> Channel: ...
8683

8784
# Create Client Credentials:
8885

@@ -378,25 +375,13 @@ class ClientCallDetails(abc.ABC):
378375
@type_check_only
379376
class _CallFuture(Call, Future[_TResponse], metaclass=abc.ABCMeta): ...
380377

381-
class UnaryUnaryClientInterceptor(abc.ABC, Generic[_TRequest, _TResponse]):
378+
class UnaryUnaryClientInterceptor(abc.ABC):
379+
# This method (not the class) is generic over _TRequest and _TResponse
380+
# and the types must satisfy the no-op implementation of
381+
# `return continuation(client_call_details, request)`.
382382
@abc.abstractmethod
383383
def intercept_unary_unary(
384384
self,
385-
# FIXME: decode these cryptic runes to confirm the typing mystery of
386-
# this callable's signature that was left for us by past civilisations:
387-
#
388-
# continuation - A function that proceeds with the invocation by
389-
# executing the next interceptor in chain or invoking the actual RPC
390-
# on the underlying Channel. It is the interceptor's responsibility
391-
# to call it if it decides to move the RPC forward. The interceptor
392-
# can use response_future = continuation(client_call_details,
393-
# request) to continue with the RPC. continuation returns an object
394-
# that is both a Call for the RPC and a Future. In the event of RPC
395-
# completion, the return Call-Future's result value will be the
396-
# response message of the RPC. Should the event terminate with non-OK
397-
# status, the returned Call-Future's exception value will be an
398-
# RpcError.
399-
#
400385
continuation: Callable[[ClientCallDetails, _TRequest], _CallFuture[_TResponse]],
401386
client_call_details: ClientCallDetails,
402387
request: _TRequest,
@@ -407,7 +392,10 @@ class _CallIterator(Call, Generic[_TResponse], metaclass=abc.ABCMeta):
407392
def __iter__(self) -> Iterator[_TResponse]: ...
408393
def __next__(self) -> _TResponse: ...
409394

410-
class UnaryStreamClientInterceptor(abc.ABC, Generic[_TRequest, _TResponse]):
395+
class UnaryStreamClientInterceptor(abc.ABC):
396+
# This method (not the class) is generic over _TRequest and _TResponse
397+
# and the types must satisfy the no-op implementation of
398+
# `return continuation(client_call_details, request)`.
411399
@abc.abstractmethod
412400
def intercept_unary_stream(
413401
self,
@@ -416,20 +404,26 @@ class UnaryStreamClientInterceptor(abc.ABC, Generic[_TRequest, _TResponse]):
416404
request: _TRequest,
417405
) -> _CallIterator[_TResponse]: ...
418406

419-
class StreamUnaryClientInterceptor(abc.ABC, Generic[_TRequest, _TResponse]):
407+
class StreamUnaryClientInterceptor(abc.ABC):
408+
# This method (not the class) is generic over _TRequest and _TResponse
409+
# and the types must satisfy the no-op implementation of
410+
# `return continuation(client_call_details, request_iterator)`.
420411
@abc.abstractmethod
421412
def intercept_stream_unary(
422413
self,
423-
continuation: Callable[[ClientCallDetails, _TRequest], _CallFuture[_TResponse]],
414+
continuation: Callable[[ClientCallDetails, Iterator[_TRequest]], _CallFuture[_TResponse]],
424415
client_call_details: ClientCallDetails,
425416
request_iterator: Iterator[_TRequest],
426417
) -> _CallFuture[_TResponse]: ...
427418

428-
class StreamStreamClientInterceptor(abc.ABC, Generic[_TRequest, _TResponse]):
419+
class StreamStreamClientInterceptor(abc.ABC):
420+
# This method (not the class) is generic over _TRequest and _TResponse
421+
# and the types must satisfy the no-op implementation of
422+
# `return continuation(client_call_details, request_iterator)`.
429423
@abc.abstractmethod
430424
def intercept_stream_stream(
431425
self,
432-
continuation: Callable[[ClientCallDetails, _TRequest], _CallIterator[_TResponse]],
426+
continuation: Callable[[ClientCallDetails, Iterator[_TRequest]], _CallIterator[_TResponse]],
433427
client_call_details: ClientCallDetails,
434428
request_iterator: Iterator[_TRequest],
435429
) -> _CallIterator[_TResponse]: ...
@@ -505,7 +499,9 @@ class ServiceRpcHandler(GenericRpcHandler, metaclass=abc.ABCMeta):
505499
# Service-Side Interceptor:
506500

507501
class ServerInterceptor(abc.ABC):
508-
# This method (not the class) is generic over _TRequest and _TResponse.
502+
# This method (not the class) is generic over _TRequest and _TResponse
503+
# and the types must satisfy the no-op implementation of
504+
# `return continuation(handler_call_details)`.
509505
@abc.abstractmethod
510506
def intercept_service(
511507
self,

‎stubs/grpcio/grpc/aio/__init__.pyi‎

Lines changed: 36 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,6 @@ class AioRpcError(RpcError):
4545

4646
# Create Client:
4747

48-
class ClientInterceptor(metaclass=abc.ABCMeta): ...
49-
5048
def insecure_channel(
5149
target: str,
5250
options: _Options | None = None,
@@ -288,7 +286,7 @@ class InterceptedUnaryUnaryCall(_InterceptedCall[_TRequest, _TResponse], metacla
288286
def __await__(self) -> Generator[Incomplete, None, _TResponse]: ...
289287
def __init__(
290288
self,
291-
interceptors: Sequence[UnaryUnaryClientInterceptor[_TRequest, _TResponse]],
289+
interceptors: Sequence[UnaryUnaryClientInterceptor],
292290
request: _TRequest,
293291
timeout: float | None,
294292
metadata: Metadata,
@@ -304,7 +302,7 @@ class InterceptedUnaryUnaryCall(_InterceptedCall[_TRequest, _TResponse], metacla
304302
# pylint: disable=too-many-arguments
305303
async def _invoke(
306304
self,
307-
interceptors: Sequence[UnaryUnaryClientInterceptor[_TRequest, _TResponse]],
305+
interceptors: Sequence[UnaryUnaryClientInterceptor],
308306
method: bytes,
309307
timeout: float | None,
310308
metadata: Metadata | None,
@@ -316,47 +314,67 @@ class InterceptedUnaryUnaryCall(_InterceptedCall[_TRequest, _TResponse], metacla
316314
) -> UnaryUnaryCall[_TRequest, _TResponse]: ...
317315
def time_remaining(self) -> float | None: ...
318316

319-
class UnaryUnaryClientInterceptor(Generic[_TRequest, _TResponse], metaclass=abc.ABCMeta):
317+
class ClientInterceptor(metaclass=abc.ABCMeta): ...
318+
319+
class UnaryUnaryClientInterceptor(ClientInterceptor, metaclass=abc.ABCMeta):
320+
# This method (not the class) is generic over _TRequest and _TResponse
321+
# and the types must satisfy the no-op implementation of
322+
# `return await continuation(client_call_details, request)`.
320323
@abc.abstractmethod
321324
async def intercept_unary_unary(
322325
self,
323-
# XXX: See equivalent function in grpc types for notes about continuation:
324-
continuation: Callable[[ClientCallDetails, _TRequest], UnaryUnaryCall[_TRequest, _TResponse]],
326+
continuation: Callable[[ClientCallDetails, _TRequest], Awaitable[UnaryUnaryCall[_TRequest, _TResponse]]],
325327
client_call_details: ClientCallDetails,
326328
request: _TRequest,
327-
) -> _TResponse: ...
329+
) -> _TResponse | UnaryUnaryCall[_TRequest, _TResponse]: ...
328330

329-
class UnaryStreamClientInterceptor(Generic[_TRequest, _TResponse], metaclass=abc.ABCMeta):
331+
class UnaryStreamClientInterceptor(ClientInterceptor, metaclass=abc.ABCMeta):
332+
# This method (not the class) is generic over _TRequest and _TResponse
333+
# and the types must satisfy the no-op implementation of
334+
# `return await continuation(client_call_details, request)`.
330335
@abc.abstractmethod
331336
async def intercept_unary_stream(
332337
self,
333-
continuation: Callable[[ClientCallDetails, _TRequest], UnaryStreamCall[_TRequest, _TResponse]],
338+
continuation: Callable[[ClientCallDetails, _TRequest], Awaitable[UnaryStreamCall[_TRequest, _TResponse]]],
334339
client_call_details: ClientCallDetails,
335340
request: _TRequest,
336-
) -> AsyncIterable[_TResponse] | UnaryStreamCall[_TRequest, _TResponse]: ...
341+
) -> AsyncIterator[_TResponse] | UnaryStreamCall[_TRequest, _TResponse]: ...
337342

338-
class StreamUnaryClientInterceptor(Generic[_TRequest, _TResponse], metaclass=abc.ABCMeta):
343+
class StreamUnaryClientInterceptor(ClientInterceptor, metaclass=abc.ABCMeta):
344+
# This method (not the class) is generic over _TRequest and _TResponse
345+
# and the types must satisfy the no-op implementation of
346+
# `return await continuation(client_call_details, request_iterator)`.
339347
@abc.abstractmethod
340348
async def intercept_stream_unary(
341349
self,
342-
continuation: Callable[[ClientCallDetails, _TRequest], StreamUnaryCall[_TRequest, _TResponse]],
350+
continuation: Callable[
351+
[ClientCallDetails, AsyncIterable[_TRequest] | Iterable[_TRequest]], Awaitable[StreamUnaryCall[_TRequest, _TResponse]]
352+
],
343353
client_call_details: ClientCallDetails,
344354
request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
345-
) -> AsyncIterable[_TResponse] | UnaryStreamCall[_TRequest, _TResponse]: ...
355+
) -> _TResponse | StreamUnaryCall[_TRequest, _TResponse]: ...
346356

347-
class StreamStreamClientInterceptor(Generic[_TRequest, _TResponse], metaclass=abc.ABCMeta):
357+
class StreamStreamClientInterceptor(ClientInterceptor, metaclass=abc.ABCMeta):
358+
# This method (not the class) is generic over _TRequest and _TResponse
359+
# and the types must satisfy the no-op implementation of
360+
# `return await continuation(client_call_details, request_iterator)`.
348361
@abc.abstractmethod
349362
async def intercept_stream_stream(
350363
self,
351-
continuation: Callable[[ClientCallDetails, _TRequest], StreamStreamCall[_TRequest, _TResponse]],
364+
continuation: Callable[
365+
[ClientCallDetails, AsyncIterable[_TRequest] | Iterable[_TRequest]],
366+
Awaitable[StreamStreamCall[_TRequest, _TResponse]],
367+
],
352368
client_call_details: ClientCallDetails,
353369
request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
354-
) -> AsyncIterable[_TResponse] | StreamStreamCall[_TRequest, _TResponse]: ...
370+
) -> AsyncIterator[_TResponse] | StreamStreamCall[_TRequest, _TResponse]: ...
355371

356372
# Server-Side Interceptor:
357373

358374
class ServerInterceptor(metaclass=abc.ABCMeta):
359-
# This method (not the class) is generic over _TRequest and _TResponse.
375+
# This method (not the class) is generic over _TRequest and _TResponse
376+
# and the types must satisfy the no-op implementation of
377+
# `return await continuation(handler_call_details)`.
360378
@abc.abstractmethod
361379
async def intercept_service(
362380
self,

0 commit comments

Comments
 (0)