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

[chalmerlowe]

Problem

Generated client libraries and transports currently lack a centralized helper in google-api-core to apply client interceptors to a gRPC channel in a clean, order-preserving manner. Without a shared helper, downstream packages must either duplicate custom interception loops or risk nested wrapper overhead.

Additionally, OpenTelemetry tracing support in client initialization requires strict experimental feature gating to prevent premature exposure of in-development capabilities.

Solution

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

1. gRPC Interceptor Utilities (google.api_core.grpc_helpers):

   • ClientInterceptor: Type alias representing client-side gRPC interceptors across unary and streaming modes.
   • apply_interceptors: Applies an optional sequence of interceptors to a grpc.Channel in a single call via grpc.intercept_channel(channel, *interceptors). Returns the original channel unmodified if interceptors 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 unit tests in test_grpc_helpers.py covering passthrough behavior and single/multiple interceptor application.
• Added unit tests in test_observability.py validating experimental feature gating (fail-fast exception when env var is missing/disabled, successful enablement when active).

Note

For Reviewers

• Passing interceptors unpacked (*interceptors) directly into grpc.intercept_channel produces a single _InterceptedChannel dispatcher rather than $N$ nested proxy channels, preserving standard execution order (first interceptor executes first on outbound requests).

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