From 447297e25118c6afa0d2078928f01520c4dd9816 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 27 Aug 2026 05:59:55 -0400 Subject: [PATCH 1/5] feat(api-core): add ClientInterceptor and apply_interceptors helper --- .../google/api_core/grpc_helpers.py | 35 ++++++- .../tests/unit/test_grpc_helpers.py | 94 ++++++++++++++++++- 2 files changed, 125 insertions(+), 4 deletions(-) diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py index 263079e7d1f7..ab944b240198 100644 --- a/packages/google-api-core/google/api_core/grpc_helpers.py +++ b/packages/google-api-core/google/api_core/grpc_helpers.py @@ -17,7 +17,7 @@ import collections import functools import warnings -from typing import Generic, Iterator, Optional, TypeVar +from typing import Generic, Iterator, Optional, Sequence, TypeVar, Union import google.auth import google.auth.credentials @@ -25,7 +25,6 @@ import google.auth.transport.requests import google.protobuf import grpc - from google.api_core import exceptions, general_helpers # The list of gRPC Callable interfaces that return iterators. @@ -34,6 +33,14 @@ # denotes the proto response type for grpc calls P = TypeVar("P") +# Type alias representing any client-side gRPC interceptor +ClientInterceptor = Union[ + grpc.UnaryUnaryClientInterceptor, + grpc.UnaryStreamClientInterceptor, + grpc.StreamUnaryClientInterceptor, + grpc.StreamStreamClientInterceptor, +] + def _patch_callable_name(callable_): """Fix-up gRPC callable attributes. @@ -419,6 +426,30 @@ def _modify_target_for_direct_path(target: str) -> str: return target +def apply_interceptors( + channel: grpc.Channel, + interceptors: Optional[Sequence[ClientInterceptor]] = None, +) -> grpc.Channel: + """Applies a sequence of interceptors to a gRPC channel. + + The interceptors are applied in the order provided, wrapping the channel + sequentially. + + Args: + channel (grpc.Channel): The channel to intercept. + interceptors (Optional[Sequence[ClientInterceptor]]): An optional sequence + of client interceptors to apply. + + Returns: + grpc.Channel: The intercepted channel, or the original channel if no + interceptors were provided. + """ + if interceptors: + for interceptor in interceptors: + channel = grpc.intercept_channel(channel, interceptor) + return channel + + _MethodCall = collections.namedtuple( "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression") ) diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py index 69281d58109b..677fc15ce2d3 100644 --- a/packages/google-api-core/tests/unit/test_grpc_helpers.py +++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py @@ -24,9 +24,8 @@ pytest.skip("No GRPC", allow_module_level=True) import google.auth.credentials -from google.longrunning import operations_pb2 - from google.api_core import exceptions, grpc_helpers +from google.longrunning import operations_pb2 def test__patch_callable_name(): @@ -932,3 +931,94 @@ def test_subscribe_unsubscribe(self): def test_close(self): channel = grpc_helpers.ChannelStub() assert channel.close() is None + + +@pytest.mark.parametrize("falsy_interceptors", [None, [], ()]) +def test_apply_interceptors_passthrough(falsy_interceptors): + """Verify that falsy or empty interceptor sequences return the channel unmodified.""" + mock_channel = mock.Mock() + result = grpc_helpers.apply_interceptors(mock_channel, falsy_interceptors) + assert result is mock_channel + + +@pytest.mark.parametrize("count", [1, 2, 3]) +def test_apply_interceptors_wrapping(count): + """Verify that interceptors are wrapped sequentially in the order provided. + + When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors + must pass the base channel and i_0 to grpc.intercept_channel, then pass the + resulting wrapped channel and i_1 to grpc.intercept_channel, and so on. + This ensures each subsequent interceptor wraps the preceding channel state. + """ + mock_channel = mock.Mock(name="base_channel") + # Generate distinct mock interceptors and the expected wrapped channel returns for each step + interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)] + wrapped_channels = [mock.Mock(name=f"wrapped_channel_{i}") for i in range(count)] + + with mock.patch( + "grpc.intercept_channel", side_effect=wrapped_channels + ) as mock_intercept: + result = grpc_helpers.apply_interceptors(mock_channel, interceptors) + + # The final return value must be the outermost wrapped channel from the final loop iteration + assert result is wrapped_channels[-1] + assert mock_intercept.call_count == count + + # Construct the expected sequential chaining: (base, i_0) -> (wrapped_0, i_1) -> ... + expected_calls = [] + current_channel = mock_channel + for i, interceptor in enumerate(interceptors): + expected_calls.append(mock.call(current_channel, interceptor)) + current_channel = wrapped_channels[i] + + mock_intercept.assert_has_calls(expected_calls) + + +def test_apply_interceptors_execution_order(): + """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel. + + In gRPC Python, sequential wrapping via grpc.intercept_channel(channel, i) creates an + 'onion' layer where the LAST applied interceptor becomes the OUTSIDE layer. + Therefore, given [i1, i2]: + - i1 wraps the raw channel (innermost layer) + - i2 wraps the result of (channel + i1) (outermost layer) + + During an RPC invocation: + 1. i2 intercepts the call first (request inbound / pre-call) + 2. i2 calls continuation(), which triggers i1 + 3. i1 calls continuation(), which reaches the channel stub / network + 4. i1 post-call logic finishes + 5. i2 post-call logic finishes + """ + execution_order = [] + + class OrderInterceptor(grpc.UnaryUnaryClientInterceptor): + def __init__(self, name): + self.name = name + + def intercept_unary_unary(self, continuation, client_call_details, request): + execution_order.append(f"{self.name}_start") + response = continuation(client_call_details, request) + execution_order.append(f"{self.name}_end") + return response + + i1 = OrderInterceptor("i1") + i2 = OrderInterceptor("i2") + + mock_channel = mock.Mock(spec=grpc.Channel) + mock_callable = mock.Mock(spec=grpc.UnaryUnaryMultiCallable) + mock_call = mock.Mock(spec=grpc.Call) + expected_response = operations_pb2.Operation(name="test_op") + mock_callable.with_call.return_value = (expected_response, mock_call) + mock_channel.unary_unary.return_value = mock_callable + + # Apply interceptors in sequence [i1, i2] + intercepted_channel = grpc_helpers.apply_interceptors(mock_channel, [i1, i2]) + + # Trigger a unary RPC through the intercepted channel + stub = operations_pb2.OperationsStub(intercepted_channel) + response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op")) + + assert response.name == "test_op" + # Verify i2 executed as the outer layer surrounding i1 + assert execution_order == ["i2_start", "i1_start", "i1_end", "i2_end"] From 593d049a1dbc01c401477b7d6382b35650828b43 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 27 Aug 2026 06:30:02 -0400 Subject: [PATCH 2/5] fix(api-core): unpack interceptors directly into grpc.intercept_channel --- .../google/api_core/grpc_helpers.py | 7 ++- .../tests/unit/test_grpc_helpers.py | 49 +++++++------------ 2 files changed, 22 insertions(+), 34 deletions(-) diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py index ab944b240198..bb2a3523d735 100644 --- a/packages/google-api-core/google/api_core/grpc_helpers.py +++ b/packages/google-api-core/google/api_core/grpc_helpers.py @@ -432,8 +432,8 @@ def apply_interceptors( ) -> grpc.Channel: """Applies a sequence of interceptors to a gRPC channel. - The interceptors are applied in the order provided, wrapping the channel - sequentially. + The interceptors are applied in the order provided, such that the first + interceptor in the sequence is the outermost layer (executes first). Args: channel (grpc.Channel): The channel to intercept. @@ -445,8 +445,7 @@ def apply_interceptors( interceptors were provided. """ if interceptors: - for interceptor in interceptors: - channel = grpc.intercept_channel(channel, interceptor) + return grpc.intercept_channel(channel, *interceptors) return channel diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py index 677fc15ce2d3..b9fc553e6555 100644 --- a/packages/google-api-core/tests/unit/test_grpc_helpers.py +++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py @@ -943,52 +943,41 @@ def test_apply_interceptors_passthrough(falsy_interceptors): @pytest.mark.parametrize("count", [1, 2, 3]) def test_apply_interceptors_wrapping(count): - """Verify that interceptors are wrapped sequentially in the order provided. + """Verify that interceptors are passed to grpc.intercept_channel in a single call. When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors - must pass the base channel and i_0 to grpc.intercept_channel, then pass the - resulting wrapped channel and i_1 to grpc.intercept_channel, and so on. - This ensures each subsequent interceptor wraps the preceding channel state. + must pass the base channel and all interceptors unpacked (*interceptors) to + grpc.intercept_channel. This creates a single intercepted channel wrapper rather than + multiple nested wrappers. """ mock_channel = mock.Mock(name="base_channel") - # Generate distinct mock interceptors and the expected wrapped channel returns for each step + mock_intercepted = mock.Mock(name="intercepted_channel") interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)] - wrapped_channels = [mock.Mock(name=f"wrapped_channel_{i}") for i in range(count)] with mock.patch( - "grpc.intercept_channel", side_effect=wrapped_channels + "grpc.intercept_channel", return_value=mock_intercepted ) as mock_intercept: result = grpc_helpers.apply_interceptors(mock_channel, interceptors) - # The final return value must be the outermost wrapped channel from the final loop iteration - assert result is wrapped_channels[-1] - assert mock_intercept.call_count == count - - # Construct the expected sequential chaining: (base, i_0) -> (wrapped_0, i_1) -> ... - expected_calls = [] - current_channel = mock_channel - for i, interceptor in enumerate(interceptors): - expected_calls.append(mock.call(current_channel, interceptor)) - current_channel = wrapped_channels[i] - - mock_intercept.assert_has_calls(expected_calls) + assert result is mock_intercepted + mock_intercept.assert_called_once_with(mock_channel, *interceptors) def test_apply_interceptors_execution_order(): """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel. - In gRPC Python, sequential wrapping via grpc.intercept_channel(channel, i) creates an - 'onion' layer where the LAST applied interceptor becomes the OUTSIDE layer. + In standard gRPC Python, grpc.intercept_channel(channel, *interceptors) processes + the interceptor list such that the first interceptor in the sequence is the outermost layer. Therefore, given [i1, i2]: - - i1 wraps the raw channel (innermost layer) - - i2 wraps the result of (channel + i1) (outermost layer) + - i1 is the outermost layer (executes first on outbound request) + - i2 is the inner layer (executes second on outbound request) During an RPC invocation: - 1. i2 intercepts the call first (request inbound / pre-call) - 2. i2 calls continuation(), which triggers i1 - 3. i1 calls continuation(), which reaches the channel stub / network - 4. i1 post-call logic finishes - 5. i2 post-call logic finishes + 1. i1 intercepts the call first (request inbound / pre-call) + 2. i1 calls continuation(), which triggers i2 + 3. i2 calls continuation(), which reaches the channel stub / network + 4. i2 post-call logic finishes + 5. i1 post-call logic finishes """ execution_order = [] @@ -1020,5 +1009,5 @@ def intercept_unary_unary(self, continuation, client_call_details, request): response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op")) assert response.name == "test_op" - # Verify i2 executed as the outer layer surrounding i1 - assert execution_order == ["i2_start", "i1_start", "i1_end", "i2_end"] + # Verify i1 executed as the outer layer surrounding i2 + assert execution_order == ["i1_start", "i2_start", "i2_end", "i1_end"] From 0e0c229d50b39f27052a8e79e3ef074498e03d5e Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 27 Aug 2026 09:26:26 -0400 Subject: [PATCH 3/5] docs(api-core): clarify apply_interceptors execution order in docstring --- packages/google-api-core/google/api_core/grpc_helpers.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py index bb2a3523d735..2f0ef9631dd8 100644 --- a/packages/google-api-core/google/api_core/grpc_helpers.py +++ b/packages/google-api-core/google/api_core/grpc_helpers.py @@ -430,10 +430,10 @@ def apply_interceptors( channel: grpc.Channel, interceptors: Optional[Sequence[ClientInterceptor]] = None, ) -> grpc.Channel: - """Applies a sequence of interceptors to a gRPC channel. + """Applies client interceptors to a gRPC channel. - The interceptors are applied in the order provided, such that the first - interceptor in the sequence is the outermost layer (executes first). + The first interceptor in the sequence is the outermost layer: it + executes first on outbound requests and last on inbound responses. Args: channel (grpc.Channel): The channel to intercept. From fba514fcad642cffff705e7b3063373fe87907b0 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 27 Aug 2026 09:45:16 -0400 Subject: [PATCH 4/5] test(api-core): remove redundant execution order test and simplify interceptor unit tests --- .../tests/unit/test_grpc_helpers.py | 55 +------------------ 1 file changed, 2 insertions(+), 53 deletions(-) diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py index b9fc553e6555..0dad58217545 100644 --- a/packages/google-api-core/tests/unit/test_grpc_helpers.py +++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py @@ -943,12 +943,11 @@ def test_apply_interceptors_passthrough(falsy_interceptors): @pytest.mark.parametrize("count", [1, 2, 3]) def test_apply_interceptors_wrapping(count): - """Verify that interceptors are passed to grpc.intercept_channel in a single call. + """Verify that interceptors are passed to grpc.intercept_channel unpacked in a single call. When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors must pass the base channel and all interceptors unpacked (*interceptors) to - grpc.intercept_channel. This creates a single intercepted channel wrapper rather than - multiple nested wrappers. + grpc.intercept_channel. """ mock_channel = mock.Mock(name="base_channel") mock_intercepted = mock.Mock(name="intercepted_channel") @@ -961,53 +960,3 @@ def test_apply_interceptors_wrapping(count): assert result is mock_intercepted mock_intercept.assert_called_once_with(mock_channel, *interceptors) - - -def test_apply_interceptors_execution_order(): - """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel. - - In standard gRPC Python, grpc.intercept_channel(channel, *interceptors) processes - the interceptor list such that the first interceptor in the sequence is the outermost layer. - Therefore, given [i1, i2]: - - i1 is the outermost layer (executes first on outbound request) - - i2 is the inner layer (executes second on outbound request) - - During an RPC invocation: - 1. i1 intercepts the call first (request inbound / pre-call) - 2. i1 calls continuation(), which triggers i2 - 3. i2 calls continuation(), which reaches the channel stub / network - 4. i2 post-call logic finishes - 5. i1 post-call logic finishes - """ - execution_order = [] - - class OrderInterceptor(grpc.UnaryUnaryClientInterceptor): - def __init__(self, name): - self.name = name - - def intercept_unary_unary(self, continuation, client_call_details, request): - execution_order.append(f"{self.name}_start") - response = continuation(client_call_details, request) - execution_order.append(f"{self.name}_end") - return response - - i1 = OrderInterceptor("i1") - i2 = OrderInterceptor("i2") - - mock_channel = mock.Mock(spec=grpc.Channel) - mock_callable = mock.Mock(spec=grpc.UnaryUnaryMultiCallable) - mock_call = mock.Mock(spec=grpc.Call) - expected_response = operations_pb2.Operation(name="test_op") - mock_callable.with_call.return_value = (expected_response, mock_call) - mock_channel.unary_unary.return_value = mock_callable - - # Apply interceptors in sequence [i1, i2] - intercepted_channel = grpc_helpers.apply_interceptors(mock_channel, [i1, i2]) - - # Trigger a unary RPC through the intercepted channel - stub = operations_pb2.OperationsStub(intercepted_channel) - response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op")) - - assert response.name == "test_op" - # Verify i1 executed as the outer layer surrounding i2 - assert execution_order == ["i1_start", "i2_start", "i2_end", "i1_end"] From 0a0f0e0741f184b80cd00051513890819c6c44c5 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 27 Aug 2026 10:08:49 -0400 Subject: [PATCH 5/5] test(api-core): align mock variable naming to Option 1 convention --- .../tests/unit/test_grpc_helpers.py | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py index 0dad58217545..9e41bab853df 100644 --- a/packages/google-api-core/tests/unit/test_grpc_helpers.py +++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py @@ -936,9 +936,9 @@ def test_close(self): @pytest.mark.parametrize("falsy_interceptors", [None, [], ()]) def test_apply_interceptors_passthrough(falsy_interceptors): """Verify that falsy or empty interceptor sequences return the channel unmodified.""" - mock_channel = mock.Mock() - result = grpc_helpers.apply_interceptors(mock_channel, falsy_interceptors) - assert result is mock_channel + mock_base_channel = mock.Mock(name="base_channel") + result = grpc_helpers.apply_interceptors(mock_base_channel, falsy_interceptors) + assert result is mock_base_channel @pytest.mark.parametrize("count", [1, 2, 3]) @@ -949,14 +949,16 @@ def test_apply_interceptors_wrapping(count): must pass the base channel and all interceptors unpacked (*interceptors) to grpc.intercept_channel. """ - mock_channel = mock.Mock(name="base_channel") - mock_intercepted = mock.Mock(name="intercepted_channel") - interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)] + mock_base_channel = mock.Mock(name="base_channel") + mock_wrapped_channel = mock.Mock(name="wrapped_channel") + mock_interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)] with mock.patch( - "grpc.intercept_channel", return_value=mock_intercepted - ) as mock_intercept: - result = grpc_helpers.apply_interceptors(mock_channel, interceptors) + "grpc.intercept_channel", return_value=mock_wrapped_channel + ) as mock_intercept_channel: + result = grpc_helpers.apply_interceptors(mock_base_channel, mock_interceptors) - assert result is mock_intercepted - mock_intercept.assert_called_once_with(mock_channel, *interceptors) + assert result is mock_wrapped_channel + mock_intercept_channel.assert_called_once_with( + mock_base_channel, *mock_interceptors + )