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 6 commits into
base: main
Choose a base branch
from
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
6 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
f10fd03
feat(api-core): update default env var to GOOGLE_SDK_EXPERIMENTAL_PYT…
chalmerlowe Aug 28, 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
2 changes: 1 addition & 1 deletion packages/google-api-core/google/api_core/_observability.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@

def is_otel_capabilities_enabled(
client_options: Optional[ClientOptions | dict[str, Any]] = None,
env_var: str = "GOOGLE_CLOUD_PYTHON_TRACING_ENABLED",
env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
) -> bool:
"""Checks if OTel capabilities are enabled and installed.

Expand Down
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 •
edited
Loading

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

❌ Not Recommended See here for full details

My response to your core point can be found at the link, but one quick bit of tangential context may help with other conversations.

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
)
43 changes: 40 additions & 3 deletions packages/google-api-core/tests/unit/test_observability.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,25 +15,27 @@
import sys
from unittest import mock

import pytest
from google.api_core import _observability
from google.api_core._feature_gating_helpers import FeatureGatingError
from google.api_core.client_options import ClientOptions


def test_is_otel_capabilities_enabled_disabled(monkeypatch):
monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "false")
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
assert not _observability.is_otel_capabilities_enabled()


def test_is_otel_capabilities_enabled_otel_missing(monkeypatch):
monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true")
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
# Simulate OTel not being installed by blocking imports
monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)

assert not _observability.is_otel_capabilities_enabled()


def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true")
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")

mock_otel = mock.Mock()
mock_otel_grpc = mock_otel.instrumentation.grpc
Expand All @@ -49,6 +51,41 @@ def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
assert _observability.is_otel_capabilities_enabled()


def test_is_otel_capabilities_enabled_experimental_requires_env_var(monkeypatch):
"""Proves that passing client_options with tracer_provider without the experimental
env var set to 'true' raises FeatureGatingError (Fail Fast).
"""
monkeypatch.delenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", raising=False)
options = ClientOptions(tracer_provider=object())

with pytest.raises(
FeatureGatingError,
match="requires GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
):
_observability.is_otel_capabilities_enabled(options)


def test_is_otel_capabilities_enabled_experimental_enabled_with_config(monkeypatch):
"""Proves that when GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED=true and tracer_provider
is supplied via client_options, is_otel_capabilities_enabled returns True.
"""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")

mock_otel = mock.Mock()
mock_otel_grpc = mock_otel.instrumentation.grpc

monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
monkeypatch.setitem(
sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
)
monkeypatch.setitem(
sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
)

options = ClientOptions(tracer_provider=object())
assert _observability.is_otel_capabilities_enabled(options)


def test_apply_otel_capabilities_to_channel_enabled_otel_installed(monkeypatch):
mock_channel = mock.Mock()
mock_intercepted_channel = mock.Mock()
Expand Down
Loading