Skip to content

feat(api-core): add ClientInterceptor and apply_interceptors helper #18236

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chalmerlowe wants to merge 5 commits into
base: main
Choose a base branch
from
+64 −4
Open

feat(api-core): add ClientInterceptor and apply_interceptors helper #18236

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
447297e
feat(api-core): add ClientInterceptor and apply_interceptors helper
chalmerlowe Aug 27, 2026
593d049
fix(api-core): unpack interceptors directly into grpc.intercept_channel
chalmerlowe Aug 27, 2026
0e0c229
docs(api-core): clarify apply_interceptors execution order in docstring
chalmerlowe Aug 27, 2026
fba514f
test(api-core): remove redundant execution order test and simplify in…
chalmerlowe Aug 27, 2026
0a0f0e0
test(api-core): align mock variable naming to Option 1 convention
chalmerlowe Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 32 additions & 2 deletions packages/google-api-core/google/api_core/grpc_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,14 @@
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
import google.auth.transport.grpc
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.
Expand All @@ -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.
Expand Down Expand Up @@ -419,6 +426,29 @@ def _modify_target_for_direct_path(target: str) -> str:
return target


def apply_interceptors(
channel: grpc.Channel,
interceptors: Optional[Sequence[ClientInterceptor]] = None,

@daniel-sanche daniel-sanche Aug 27, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In your other PR, I suggested accepting Callable wrappers here as an alternate interceptor type, to support the otel interceptor.

That would make this into something like:

modified_channel = channel
for interceptor in interceptors or []:
    if isinstance(interceptor, ClientInterceptor):
        modified_channel = grpc.intercept_channel(channel, interceptor)
    else:
        modified_channel = interceptor(modified_channel)
return modified channel

Let me know if you think that could work

@chalmerlowe chalmerlowe Aug 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@daniel-sanche

I am thinking this through and will respond to your core point, but one quick bit of tangential context for now.

for loops wrap in reverse order compared to *interceptors

If we use a for loop, we have to adapt it here to align with how grpc.intercept_channel() works internally.

grpc.intercept_channel(..., *interceptors) unpacks and reverses the list of interceptors it receives. Thus a straight up for loop like this does not account for that and wraps the channel in the wrong order.

The full code is below, but this is the relevant line from the grpc.intercept_channels() function:

for interceptor in reversed(list(interceptors)):

Thus, if we want to build out a channel via for loop, we have to make sure the interceptors we feed in are in the same order that the grpc.intercept_channel() function would expect them to be. The proposed version behaves thus:

interceptors = [1, 2, 3, 4]
for i in interceptors:
    modified_channel = grpc.intercept_channel(channel, i)

yields something akin to this:

4(3(2(1(channel))))

But a straight call to grpc.intercept_channel(channel, *interceptors)
is handled in the following way internally:

    reversed_list = reversed(list(interceptors)) # [1, 2, 3, 4] becomes [4, 3, 2, 1]
    for i in reversed_list:
        channel = _Channel(channel, interceptor)
    return channel

and yields:

1(2(3(4(channel))))

Code from grpc package:

def intercept_channel(
    channel: grpc.Channel,
    *interceptors: Optional[
        Sequence[
            Union[
                grpc.UnaryUnaryClientInterceptor,
                grpc.UnaryStreamClientInterceptor,
                grpc.StreamStreamClientInterceptor,
                grpc.StreamUnaryClientInterceptor,
            ]
        ]
    ],
) -> grpc.Channel:
    for interceptor in reversed(list(interceptors)):
        if (
            not isinstance(interceptor, grpc.UnaryUnaryClientInterceptor)
            and not isinstance(interceptor, grpc.UnaryStreamClientInterceptor)
            and not isinstance(interceptor, grpc.StreamUnaryClientInterceptor)
            and not isinstance(interceptor, grpc.StreamStreamClientInterceptor)
        ):
            error_msg = (
                "interceptor must be "
                "grpc.UnaryUnaryClientInterceptor or "
                "grpc.UnaryStreamClientInterceptor or "
                "grpc.StreamUnaryClientInterceptor or "
                "grpc.StreamStreamClientInterceptor"
            )
            raise TypeError(error_msg)
        channel = _Channel(channel, interceptor)
    return channel
``

@chalmerlowe chalmerlowe Aug 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In your other PR, I suggested accepting Callable wrappers here as an alternate interceptor type, to support the otel interceptor.

That would make this into something like:

modified_channel = channel
for interceptor in interceptors or []:
    if isinstance(interceptor, ClientInterceptor):
        modified_channel = grpc.intercept_channel(channel, interceptor)
    else:
        modified_channel = interceptor(modified_channel)
return modified channel

Let me know if you think that could work

Note my longer reply elsewhere in PR 18188 about why I don't think this is a good idea: basically this breaks separation of concerns and introduces multiple intermediary complications.

) -> grpc.Channel:
"""Applies client interceptors to a gRPC channel.

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.
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:
return grpc.intercept_channel(channel, *interceptors)
return channel
Comment thread
chalmerlowe marked this conversation as resolved.


_MethodCall = collections.namedtuple(
"_MethodCall", ("request", "timeout", "metadata", "credentials", "compression")
)
Expand Down
34 changes: 32 additions & 2 deletions packages/google-api-core/tests/unit/test_grpc_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -932,3 +931,34 @@ 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_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])
def test_apply_interceptors_wrapping(count):
"""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.
"""
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_wrapped_channel
) as mock_intercept_channel:
result = grpc_helpers.apply_interceptors(mock_base_channel, mock_interceptors)

assert result is mock_wrapped_channel
mock_intercept_channel.assert_called_once_with(
mock_base_channel, *mock_interceptors
)
Loading