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

packages/google-api-core/google/api_core/_observability.py

@@ -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.
 

packages/google-api-core/google/api_core/grpc_helpers.py

@@ -17,7 +17,17 @@
 import collections
 import functools
 import warnings
-from typing import Generic, Iterator, Optional, TypeVar
+from typing import (
+    Callable,
+    Generic,
+    Iterator,
+    Optional,
+    Sequence,
+    TypeAlias,
+    TypeVar,
+    cast,
+    get_args,
+)
 
 import google.auth
 import google.auth.credentials
@@ -34,6 +44,23 @@
 # denotes the proto response type for grpc calls
 P = TypeVar("P")
 
+# Type alias representing any client-side gRPC interceptor
+ClientInterceptor: TypeAlias = (
+    grpc.UnaryUnaryClientInterceptor
+    | grpc.UnaryStreamClientInterceptor
+    | grpc.StreamUnaryClientInterceptor
+    | grpc.StreamStreamClientInterceptor
+)
+
+# Runtime tuple of gRPC client interceptor base classes for isinstance checks
+_CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
+
+# Type alias representing a channel-wrapping callable
+ChannelWrapperCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel]
+
+# Generic type alias representing any channel wrapper (interceptor or callable)
+ChannelWrapper: TypeAlias = ClientInterceptor | ChannelWrapperCallable
+
 
 def _patch_callable_name(callable_):
     """Fix-up gRPC callable attributes.
@@ -419,6 +446,48 @@ def _modify_target_for_direct_path(target: str) -> str:
     return target
 
 
+def apply_channel_wrappers(
+    channel: grpc.Channel,
+    wrappers: Sequence[ChannelWrapper] | None = None,
+) -> grpc.Channel:
+    """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel.
+
+    Executes in reverse order so the first wrapper in the sequence becomes the
+    outermost layer on outbound requests and the innermost layer on inbound responses.
+
+    Args:
+        channel (grpc.Channel): The channel to wrap.
+        wrappers (Optional[Sequence[ChannelWrapper]]):
+            An optional sequence of client interceptors or channel-wrapping
+            callables to apply.
+
+    Returns:
+        grpc.Channel: The wrapped channel, or the original channel if no
+            wrappers were provided.
+
+    Raises:
+        TypeError: If an item in ``wrappers`` is neither a gRPC ClientInterceptor
+            nor a Callable[[Channel], Channel].
+    """
+    if not wrappers:
+        return channel
+
+    modified_channel = channel
+    # Reverse the inputs to align with the behavior of grpc.create_channel(*interceptors)
+    for wrapper in reversed(list(wrappers)):
+        if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
+            modified_channel = grpc.intercept_channel(modified_channel, wrapper)
+        elif callable(wrapper):
+            wrapper_callable = cast(ChannelWrapperCallable, wrapper)
+            modified_channel = wrapper_callable(modified_channel)
+        else:
+            raise TypeError(
+                f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}"
+            )
+
+    return modified_channel
+
+
 _MethodCall = collections.namedtuple(
     "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression")
 )

packages/google-api-core/tests/unit/test_client_options.py

@@ -15,6 +15,7 @@
 from re import match
 
 import pytest
+
 from google.api_core import client_options
 
 from ..helpers import warn_deprecated_credentials_file

packages/google-api-core/tests/unit/test_grpc_helpers.py

@@ -932,3 +932,102 @@ def test_subscribe_unsubscribe(self):
     def test_close(self):
         channel = grpc_helpers.ChannelStub()
         assert channel.close() is None
+
+
+@pytest.mark.parametrize("falsy_wrappers", [None, [], ()])
+def test_apply_channel_wrappers_passthrough(falsy_wrappers):
+    """Verify that falsy or empty wrapper sequences return the channel unmodified."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    result = grpc_helpers.apply_channel_wrappers(mock_base_channel, falsy_wrappers)
+    assert result is mock_base_channel
+
+
+def test_apply_channel_wrappers_grpc_client_interceptors():
+    """Verify that standard gRPC ClientInterceptor instances are applied via grpc.intercept_channel."""
+
+    class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            return continuation(client_call_details, request)
+
+    class DummyStreamInterceptor(grpc.StreamStreamClientInterceptor):
+        def intercept_stream_stream(
+            self, continuation, client_call_details, request_iterator
+        ):
+            return continuation(client_call_details, request_iterator)
+
+    interceptor1 = DummyUnaryInterceptor()
+    interceptor2 = DummyStreamInterceptor()
+
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_after_i2 = mock.Mock(name="chan_after_i2")
+    mock_chan_after_i1 = mock.Mock(name="chan_after_i1")
+
+    with mock.patch(
+        "grpc.intercept_channel",
+        side_effect=[mock_chan_after_i2, mock_chan_after_i1],
+    ) as mock_intercept:
+        result = grpc_helpers.apply_channel_wrappers(
+            mock_base_channel, [interceptor1, interceptor2]
+        )
+
+    assert result is mock_chan_after_i1
+    assert mock_intercept.call_count == 2
+    # Executed in reverse order so interceptor1 is outermost
+    mock_intercept.assert_has_calls(
+        [
+            mock.call(mock_base_channel, interceptor2),
+            mock.call(mock_chan_after_i2, interceptor1),
+        ]
+    )
+
+
+def test_apply_channel_wrappers_callables():
+    """Verify that channel-wrapping callables Callable[[Channel], Channel] are invoked in sequence."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_1 = mock.Mock(name="chan_1")
+    mock_chan_2 = mock.Mock(name="chan_2")
+
+    wrapper1 = mock.Mock(side_effect=lambda ch: mock_chan_2)
+    wrapper2 = mock.Mock(side_effect=lambda ch: mock_chan_1)
+
+    result = grpc_helpers.apply_channel_wrappers(
+        mock_base_channel, [wrapper1, wrapper2]
+    )
+
+    assert result is mock_chan_2
+    # Executed in reverse order: wrapper2 runs first on base channel, then wrapper1
+    wrapper2.assert_called_once_with(mock_base_channel)
+    wrapper1.assert_called_once_with(mock_chan_1)
+
+
+def test_apply_channel_wrappers_interspersed():
+    """Verify that a mixed sequence of gRPC interceptors and channel wrapper callables are applied."""
+
+    class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            return continuation(client_call_details, request)
+
+    interceptor = DummyUnaryInterceptor()
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_after_wrapper = mock.Mock(name="chan_after_wrapper")
+    mock_chan_after_interceptor = mock.Mock(name="chan_after_interceptor")
+
+    wrapper = mock.Mock(return_value=mock_chan_after_wrapper)
+
+    with mock.patch(
+        "grpc.intercept_channel", return_value=mock_chan_after_interceptor
+    ) as mock_intercept:
+        result = grpc_helpers.apply_channel_wrappers(
+            mock_base_channel, [interceptor, wrapper]
+        )
+
+    assert result is mock_chan_after_interceptor
+    wrapper.assert_called_once_with(mock_base_channel)
+    mock_intercept.assert_called_once_with(mock_chan_after_wrapper, interceptor)
+
+
+def test_apply_channel_wrappers_invalid_type_raises():
+    """Verify that passing an invalid object that is neither an interceptor nor callable raises TypeError."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    with pytest.raises(TypeError, match="Expected ChannelWrapper"):
+        grpc_helpers.apply_channel_wrappers(mock_base_channel, [12345])

packages/google-api-core/tests/unit/test_observability.py

@@ -15,25 +15,28 @@
 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
@@ -49,6 +52,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()

packages/google-api-core/tests/unit/test_path_template.py

@@ -18,6 +18,7 @@
 
 import pytest
 from google.api import auth_pb2
+
 from google.api_core import path_template