refactor(api-core): simplify OTel interceptor helpers and extract tracer provider (#18263)

## Summary
Refactors `google.api_core._observability` to simplify OpenTelemetry
interceptor instantiation and use concrete return types:

- Extracts `_get_tracer_provider(client_options)` to centralize tracer
provider lookup across `ClientOptions` objects and dictionaries.
- Removes the private `_get_otel_interceptor(is_async: bool)` helper,
allowing `get_otel_interceptor` and `get_otel_async_interceptor` to
directly invoke synchronous and asynchronous OpenTelemetry APIs with
concrete return types.
- Adds dedicated unit tests for `_get_tracer_provider` and maintains
symmetrical tests for sync and async interceptor getters.

Addresses review feedback on #18236.

Author: chalmerlowe

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

@@ -24,8 +24,12 @@
 from google.api_core.client_options import ClientOptions
 
 if TYPE_CHECKING:
-    # flake8: grpc is imported only for static analysis and type annotations
+    # flake8: grpc, trace, and ClientInterceptor are imported only for static analysis and type annotations
+    # The `# noqa: F401` comment avoids flake8 "imported but not used" errors.
     import grpc  # noqa: F401
+    import opentelemetry.trace  # noqa: F401
+
+    from google.api_core.grpc_helpers import ClientInterceptor  # noqa: F401
 
 _TRACER_PROVIDER = "tracer_provider"
 
@@ -60,31 +64,23 @@ def is_otel_capabilities_enabled(
     return False
 
 
-def _get_otel_interceptor(
+def _get_tracer_provider(
     client_options: ClientOptions | dict[str, Any] | None = None,
-    is_async: bool = False,
-) -> Any:
-    """Instantiates a sync or async OpenTelemetry gRPC client interceptor.
+) -> opentelemetry.trace.TracerProvider | None:
+    """Extracts the OpenTelemetry tracer provider from client options if present.
 
     Args:
         client_options: The client options object or dictionary.
-        is_async: If True, returns an async interceptor (`aio_client_interceptors`),
-            otherwise returns a sync interceptor (`client_interceptor`).
 
     Returns:
-        Any: The instantiated OpenTelemetry client interceptor.
+        opentelemetry.trace.TracerProvider | None: The tracer provider if present,
+            None otherwise.
     """
-    import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
-
-    tracer_provider = None
     if isinstance(client_options, dict):
-        tracer_provider = client_options.get(_TRACER_PROVIDER)
+        return client_options.get(_TRACER_PROVIDER)
     elif client_options is not None:
-        tracer_provider = getattr(client_options, _TRACER_PROVIDER, None)
-
-    if is_async:
-        return otel_grpc.aio_client_interceptors(tracer_provider=tracer_provider)
-    return otel_grpc.client_interceptor(tracer_provider=tracer_provider)
+        return getattr(client_options, _TRACER_PROVIDER, None)
+    return None
 
 
 def get_otel_interceptor(
@@ -97,15 +93,17 @@ def get_otel_interceptor(
             and extracting the tracer provider.
 
     Returns:
-        Optional[Callable[[grpc.Channel], grpc.Channel]]: An interceptor callable if OpenTelemetry
+        Callable[[grpc.Channel], grpc.Channel] | None: An interceptor callable if OpenTelemetry
             tracing is enabled and installed, None otherwise.
     """
     if not is_otel_capabilities_enabled(client_options):
         return None
 
     import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
 
-    interceptor = _get_otel_interceptor(client_options, is_async=False)
+    interceptor: ClientInterceptor = otel_grpc.client_interceptor(
+        tracer_provider=_get_tracer_provider(client_options)
+    )
 
     def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
         return otel_grpc.intercept_channel(channel, interceptor)
@@ -123,10 +121,15 @@ def get_otel_async_interceptor(
             and extracting the tracer provider.
 
     Returns:
-        Optional[Sequence[grpc.aio.ClientInterceptor]]: Instantiated OpenTelemetry async
+        Sequence[grpc.aio.ClientInterceptor] | None: Instantiated OpenTelemetry async
             client interceptors if tracing is enabled and installed, None otherwise.
     """
     if not is_otel_capabilities_enabled(client_options):
         return None
 
-    return _get_otel_interceptor(client_options, is_async=True)
+    # Ignored by mypy: Optional dependency only loaded if early-return is skipped
+    import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
+
+    return otel_grpc.aio_client_interceptors(
+        tracer_provider=_get_tracer_provider(client_options)
+    )

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

@@ -23,11 +23,15 @@
 
 
 def test_is_otel_capabilities_enabled_disabled(monkeypatch):
+    """Proves that is_otel_capabilities_enabled returns False when the tracing environment variable is disabled."""
     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):
+    """Proves that is_otel_capabilities_enabled returns False when tracing is enabled
+    but OpenTelemetry gRPC instrumentation is not installed.
+    """
     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)
@@ -36,6 +40,9 @@ def test_is_otel_capabilities_enabled_otel_missing(monkeypatch):
 
 
 def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
+    """Proves that is_otel_capabilities_enabled returns True when tracing is enabled
+    and OpenTelemetry gRPC instrumentation is installed.
+    """
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
 
     mock_otel = mock.Mock()
@@ -57,7 +64,7 @@ def test_is_otel_capabilities_enabled_experimental_requires_env_var(monkeypatch)
     env var set to 'true' raises FeatureGatingError (Fail Fast).
     """
     monkeypatch.delenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", raising=False)
-    options = ClientOptions(tracer_provider=object())
+    options = ClientOptions(tracer_provider=mock.Mock())
 
     with pytest.raises(
         FeatureGatingError,
@@ -83,115 +90,54 @@ def test_is_otel_capabilities_enabled_experimental_enabled_with_config(monkeypat
         sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
     )
 
-    options = ClientOptions(tracer_provider=object())
+    options = ClientOptions(tracer_provider=mock.Mock())
     assert _observability.is_otel_capabilities_enabled(options)
 
 
-def test_get_otel_interceptor_sync_default(monkeypatch):
-    mock_otel = mock.Mock()
-    mock_otel_grpc = mock_otel.instrumentation.grpc
-    mock_interceptor = mock.Mock()
-    mock_otel_grpc.client_interceptor.return_value = mock_interceptor
-
-    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
-    )
-
-    result = _observability._get_otel_interceptor()
-    assert result is mock_interceptor
-    mock_otel_grpc.client_interceptor.assert_called_once_with(tracer_provider=None)
+def test_get_tracer_provider_default():
+    """Proves that _get_tracer_provider returns None when no client_options are supplied."""
+    assert _observability._get_tracer_provider() is None
 
 
-def test_get_otel_interceptor_sync_config(monkeypatch):
-    mock_tracer_provider = object()
+def test_get_tracer_provider_config():
+    """Proves that _get_tracer_provider extracts the tracer provider when supplied
+    via a ClientOptions instance.
+    """
+    mock_tracer_provider = mock.Mock()
     options = ClientOptions(tracer_provider=mock_tracer_provider)
+    assert _observability._get_tracer_provider(options) is mock_tracer_provider
 
-    mock_otel = mock.Mock()
-    mock_otel_grpc = mock_otel.instrumentation.grpc
-    mock_interceptor = mock.Mock()
-    mock_otel_grpc.client_interceptor.return_value = mock_interceptor
-
-    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
-    )
-
-    result = _observability._get_otel_interceptor(client_options=options)
-    assert result is mock_interceptor
-    mock_otel_grpc.client_interceptor.assert_called_once_with(
-        tracer_provider=mock_tracer_provider
-    )
 
-
-def test_get_otel_interceptor_sync_dict_config(monkeypatch):
-    mock_tracer_provider = object()
+def test_get_tracer_provider_dict_config():
+    """Proves that _get_tracer_provider extracts the tracer provider when supplied
+    via a dictionary configuration.
+    """
+    mock_tracer_provider = mock.Mock()
     options = {"tracer_provider": mock_tracer_provider}
-
-    mock_otel = mock.Mock()
-    mock_otel_grpc = mock_otel.instrumentation.grpc
-    mock_interceptor = mock.Mock()
-    mock_otel_grpc.client_interceptor.return_value = mock_interceptor
-
-    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
-    )
-
-    result = _observability._get_otel_interceptor(client_options=options)
-    assert result is mock_interceptor
-    mock_otel_grpc.client_interceptor.assert_called_once_with(
-        tracer_provider=mock_tracer_provider
-    )
-
-
-def test_get_otel_interceptor_async(monkeypatch):
-    mock_tracer_provider = object()
-    options = ClientOptions(tracer_provider=mock_tracer_provider)
-
-    mock_otel = mock.Mock()
-    mock_otel_grpc = mock_otel.instrumentation.grpc
-    mock_async_interceptors = [mock.Mock()]
-    mock_otel_grpc.aio_client_interceptors.return_value = mock_async_interceptors
-
-    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
-    )
-
-    result = _observability._get_otel_interceptor(client_options=options, is_async=True)
-    assert result is mock_async_interceptors
-    mock_otel_grpc.aio_client_interceptors.assert_called_once_with(
-        tracer_provider=mock_tracer_provider
-    )
+    assert _observability._get_tracer_provider(options) is mock_tracer_provider
 
 
 def test_get_otel_interceptor_disabled(monkeypatch):
+    """Proves that get_otel_interceptor returns None when tracing is disabled."""
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
     assert _observability.get_otel_interceptor() is None
 
 
 def test_get_otel_interceptor_otel_missing(monkeypatch):
+    """Proves that get_otel_interceptor returns None when OpenTelemetry gRPC
+    instrumentation is not installed.
+    """
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
     assert _observability.get_otel_interceptor() is None
 
 
 def test_get_otel_interceptor_enabled(monkeypatch):
+    """Proves that get_otel_interceptor creates a synchronous OpenTelemetry client
+    interceptor with the resolved tracer provider and returns a channel-intercepting callable.
+    """
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
-    mock_tracer_provider = object()
+    mock_tracer_provider = mock.Mock()
     options = ClientOptions(tracer_provider=mock_tracer_provider)
 
     mock_raw_channel = mock.Mock(name="raw_channel")
@@ -232,7 +178,7 @@ def test_get_otel_interceptor_with_apply_channel_interceptors(monkeypatch):
     from google.api_core import grpc_helpers
 
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
-    mock_tracer_provider = object()
+    mock_tracer_provider = mock.Mock()
     options = ClientOptions(tracer_provider=mock_tracer_provider)
 
     mock_raw_channel = mock.Mock(name="raw_channel")
@@ -266,19 +212,26 @@ def test_get_otel_interceptor_with_apply_channel_interceptors(monkeypatch):
 
 
 def test_get_otel_async_interceptor_disabled(monkeypatch):
+    """Proves that get_otel_async_interceptor returns None when tracing is disabled."""
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
     assert _observability.get_otel_async_interceptor() is None
 
 
 def test_get_otel_async_interceptor_otel_missing(monkeypatch):
+    """Proves that get_otel_async_interceptor returns None when OpenTelemetry gRPC
+    instrumentation is not installed.
+    """
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
     assert _observability.get_otel_async_interceptor() is None
 
 
 def test_get_otel_async_interceptor_enabled(monkeypatch):
+    """Proves that get_otel_async_interceptor instantiates and returns asynchronous
+    OpenTelemetry client interceptors with the resolved tracer provider.
+    """
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
-    mock_tracer_provider = object()
+    mock_tracer_provider = mock.Mock()
     options = ClientOptions(tracer_provider=mock_tracer_provider)
 
     mock_async_interceptors = [mock.Mock(name="otel_async_interceptor")]