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

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

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.

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

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