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

[chalmerlowe]

Problem

Generated client libraries and transports need a centralized, maintainable way in google-api-core to apply gRPC channel wrappers (standard client interceptors as well as custom channel-wrapping callables) in a clean, order-preserving pipeline. Without a shared helper, downstream packages must duplicate wrapping loops or risk wrapper ordering issues.

Because synchronous gRPC channels can be wrapped post-creation while asynchronous gRPC channels require interceptors at channel creation time, client transports need helpers that return the appropriate channel wrapper or async interceptors without duplicating OpenTelemetry resolution logic across client libraries.

Additionally, OpenTelemetry tracing support in client initialization defaults to the EXPERIMENTAL token for experimental feature gating to prevent premature invocation of in-development capabilities.

Solution

This PR introduces the following foundational utilities to google-api-core:

1. gRPC Channel Wrapper Utilities (google.api_core.grpc_helpers):

   • ClientInterceptor: Type alias representing any client-side gRPC interceptor (UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor, StreamUnaryClientInterceptor, StreamStreamClientInterceptor).
   • ChannelWrapperCallable: Type alias representing a channel-wrapping callable (Callable[[grpc.Channel], grpc.Channel]).
   • ChannelWrapper: Generic union type alias representing any channel wrapper (Union[ClientInterceptor, ChannelWrapperCallable]).
   • apply_channel_wrappers: Applies an optional sequence of channel wrappers to a grpc.Channel in reverse order so the first item in the sequence becomes the outermost layer on outbound requests. Returns the original channel unmodified if wrappers is None or empty.
   • get_otel_channel_wrapper(client_options): Returns a channel-wrapping function (Callable[[Channel], Channel]) for synchronous gRPC channels when OpenTelemetry tracing is enabled and installed.
   • get_otel_async_interceptor(client_options): Returns a list of OpenTelemetry asynchronous client interceptors (aio_client_interceptors) for use when constructing grpc.aio channels.
   • Integrates with grpc_helpers.apply_channel_wrappers to wrap raw channels using OpenTelemetry's intercept_channel.
   • _get_otel_interceptor(client_options, is_async): Internal helper that extracts tracer_provider from ClientOptions and creates the appropriate OpenTelemetry sync or async interceptors.

2. Experimental Feature Gating for Tracing (google.api_core._observability):

   • Sets the default environment variable in is_otel_capabilities_enabled to GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED.
   • Enforces fail-fast behavior: attempting to configure tracing via ClientOptions.tracer_provider without setting the experimental environment variable raises FeatureGatingError.

Testing

• Added comprehensive unit tests in test_grpc_helpers.py covering passthrough behavior, pure ClientInterceptor sequences, pure callable sequences, interspersed [interceptor, callable] sequences, and TypeError validation on invalid types.
• Added unit tests in test_observability.py validating experimental feature gating (FeatureGatingError fail-fast validation and successful enablement).
• Added unit tests in tests/unit/test_observability.py covering:
  • Sync and async interceptor extraction and tracer_provider configuration.
  • get_otel_channel_wrapper behavior when tracing is disabled, when OpenTelemetry is not installed, and when tracing is enabled.
  • Integration between get_otel_channel_wrapper and grpc_helpers.apply_channel_wrappers.
  • get_otel_async_interceptor behavior across disabled, missing, and enabled states.

Notes for Reviewers

• get_otel_channel_wrapper returns a callable rather than modifying the channel immediately, allowing transport layers to combine OpenTelemetry wrapping with user-supplied custom channel wrappers.

[gemini-code-assist]

Code Review

This pull request introduces the apply_interceptors helper function to sequentially apply a list of client interceptors to a gRPC channel, along with comprehensive unit tests verifying its behavior. The reviewer feedback correctly points out that applying interceptors sequentially in a loop introduces unnecessary nesting overhead and reverses the standard gRPC execution order. To resolve this, the reviewer suggests unpacking the interceptors directly into a single grpc.intercept_channel call and updating the corresponding execution order test assertion.

[daniel-sanche]

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]

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

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.

[chalmerlowe]

@daniel-sanche

Done.

[chalmerlowe]

Note

Blank line(s) introduced by ruff and "import order" sorting process.

[chalmerlowe]

As noted previously. Happy to revisit these in a fast-follow PR to apply fixtures, reusable functions, and/or parametrizations to reduce the size/complexity of the test suite.

Given a preference, would like to ensure this gets merged before coming back to invest heavily in what might otherwise be premature optimization.

[daniel-sanche]

LGTM, but if it's possible to improve the types before merging, that would be great

[daniel-sanche]

nit: Personally, I think over-using aliases can make things harder to read for users. They have to go look-up what ClientInterceptorType means, and then look up ClientInterceptorCallable

I think ClientInterceptor | Callable[[grpc.Channel], grpc.Channel] is more helpful

But it's probably fine either way

[daniel-sanche]

It seems a little strange that the type annotation says ClientInterceptorCallable, but the returned Callable is annotated as Callable[[Any], Any]. Does this pass the type checks?

[daniel-sanche]

If we're adding the async code here, we should probably add matching ClientInterceptorAsync type aliases too

[daniel-sanche]

Is it possible to use a concrete type here?