feat: Add Feature Gating configuration helpers.

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

@@ -0,0 +1,124 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Observability environment variable and client options resolution helpers."""
+
+import os
+import warnings
+from typing import Any, Dict, Optional, Union
+
+# Allowed truthy and falsy patterns for environment variables
+_TRUTHY_VALUES = ("y", "yes", "t", "true", "on", "1")
+_FALSY_VALUES = ("n", "no", "f", "false", "off", "0")
+
+
+def _strtobool(val: str) -> Optional[bool]:
+    """Convert a string representation of truth to a boolean."""
+    clean_val = val.lower().strip()
+    if not clean_val:
+        return None
+    if clean_val in _TRUTHY_VALUES:
+        return True
+    if clean_val in _FALSY_VALUES:
+        return False
+    raise ValueError(f"Invalid truth value: {val!r}")
+
+
+def _get_env_bool(name: str) -> Optional[bool]:
+    """Retrieve the boolean value of an environment variable."""
+    val = os.getenv(name)
+    if val is None:
+        return None
+    try:
+        return _strtobool(val)
+    except ValueError as e:
+        warnings.warn(f"Ignored invalid value for {name}: {e}", RuntimeWarning)
+        return None
+
+
+def _has_provider(
+    client_options: Optional[Union[Dict[str, Any], Any]], provider_key: str
+) -> bool:
+    """Checks if a specific provider key is present and not None in client_options."""
+    if client_options is None:
+        return False
+
+    if isinstance(client_options, dict):
+        return client_options.get(provider_key) is not None
+
+    return getattr(client_options, provider_key, None) is not None
+
+
+def resolve_feature_flags(
+    env_var: str,
+    provider_key: str,
+    client_options: Optional[Union[Dict[str, Any], Any]] = None,
+) -> bool:
+    """Determines if a feature is enabled based on environment variables and client options.
+
+    Behavior depends on whether the `env_var` name contains "EXPERIMENTAL":
+
+    - **Experimental Path** (env_var contains "EXPERIMENTAL"):
+      Strict control. Requires the environment variable to be explicitly 'true'.
+      If a programmatic provider is passed but the environment variable is not 'true',
+      raises ValueError (Fail Fast).
+
+    - **GA Path** (env_var does not contain "EXPERIMENTAL"):
+      Standard precedence. Enabled if a programmatic provider is passed,
+      otherwise falls back to the environment variable value.
+
+    Args:
+        env_var: The name of the environment variable controlling this feature.
+        provider_key: The key in client_options/attributes for the programmatic provider.
+        client_options: Optional. A dictionary or object containing client configuration.
+
+    Returns:
+        bool: True if the feature is resolved to enabled, False otherwise.
+
+    Raises:
+        ValueError: If a provider is provided for an experimental feature without enabling the experimental environment variable.
+    """
+
+    # Check for programmatic feature provider
+    has_provider = _has_provider(client_options, provider_key)
+
+    # Read environment variable
+    env_var_setting = _get_env_bool(env_var)
+
+    # EXPERIMENTAL PATH:
+    # Resolution Hierarchy:
+    #   1. EXPERIMENTAL Environment Variable
+    #   2. Fail Fast if Provider present but EXPERIMENTAL Environment Variable is not enabled
+    if "EXPERIMENTAL" in env_var:
+        # Fail Fast if provider present but experimental environment variable is not enabled
+        if env_var_setting is not True and has_provider:
+            raise ValueError(
+                f"Experimental feature requires {env_var} to be set to 'true' to use programmatic providers."
+            )
+
+        return bool(env_var_setting)
+
+    # GENERAL AVAILABILITY PATH:
+    # Resolution Hierarchy:
+    #   1. Programmatic Provider
+    #   2. Environment Variable
+
+    # Check Programmatic Provider
+    if has_provider:
+        return True
+
+    # Check Environment Variable
+    return bool(env_var_setting)

packages/google-api-core/pyproject.toml

@@ -94,4 +94,6 @@ filterwarnings = [
   "ignore:.*custom tp_new.*in Python 3.14:DeprecationWarning",
   # Remove once https://github.com/grpc/grpc/issues/35086 is fixed (and version newer than 1.60.0 is published)
   "ignore:There is no current event loop:DeprecationWarning",
+  # Ignore external OpenTelemetry/importlib.metadata SelectableGroups warning
+  "ignore:.*SelectableGroups dict interface is deprecated:DeprecationWarning",
 ]

packages/google-api-core/testing/constraints-3.10.txt

@@ -12,3 +12,4 @@ requests==2.33.0
 grpcio==1.41.0
 grpcio-status==1.41.0
 proto-plus==1.24.0
+opentelemetry-api==1.27.0

packages/google-api-core/testing/constraints-async-rest-3.10.txt

@@ -13,3 +13,4 @@ grpcio==1.41.0
 grpcio-status==1.41.0
 proto-plus==1.24.0
 aiohttp==3.13.4
+opentelemetry-api==1.27.0

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

@@ -0,0 +1,199 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+from google.api_core import feature_gating_helpers
+from google.api_core.feature_gating_helpers import (
+    _get_env_bool,
+    _strtobool,
+)
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        # Truthy values
+        ("y", True),
+        ("yes", True),
+        ("t", True),
+        ("true", True),
+        ("on", True),
+        ("1", True),
+        ("  True  ", True),
+        # Falsy values
+        ("n", False),
+        ("no", False),
+        ("f", False),
+        ("false", False),
+        ("off", False),
+        ("0", False),
+        (" FALSE ", False),
+        # Empty string
+        ("", None),
+    ],
+)
+def test_strtobool(value, expected):
+    assert _strtobool(value) is expected
+
+
+def test_strtobool_invalid():
+    with pytest.raises(ValueError):
+        _strtobool("invalid")
+
+
+def test_get_env_bool(monkeypatch):
+    monkeypatch.setenv("TEST_VAR", "true")
+    assert _get_env_bool("TEST_VAR") is True
+
+    monkeypatch.setenv("TEST_VAR", "invalid")
+    import pytest
+
+    with pytest.warns(RuntimeWarning, match="Ignored invalid value"):
+        assert _get_env_bool("TEST_VAR") is None
+
+    monkeypatch.delenv("TEST_VAR", raising=False)
+    assert _get_env_bool("TEST_VAR") is None
+
+
+def test_resolve_feature_flags_ga_enabled_via_env(monkeypatch):
+    """Verify that a GA feature is enabled if its environment variable is True."""
+    # Setup: We pass a GA environment variable set to True
+    monkeypatch.setenv("GOOGLE_SDK_PYTHON_TRACING_ENABLED", "true")
+
+    # Action
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=None,
+    )
+
+    # Assertion
+    assert result is True
+
+
+@pytest.mark.parametrize("exp_env_state", [None, "false"], ids=["missing", "disabled"])
+def test_resolve_feature_flags_exp_blocked_with_provider_fails_fast(
+    monkeypatch, exp_env_state
+):
+    """Verify that passing a provider to an experimental feature raises ValueError if the experimental environment variable is disabled or missing."""
+    # Setup: Experimental env var is set to exp_env_state (None means not set)
+    if exp_env_state is not None:
+        monkeypatch.setenv(
+            "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", exp_env_state
+        )
+    else:
+        monkeypatch.delenv(
+            "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", raising=False
+        )
+    client_options = {"tracer_provider": object()}
+
+    # Action & Assertion
+    with pytest.raises(ValueError, match="Experimental feature"):
+        feature_gating_helpers.resolve_feature_flags(
+            env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+            provider_key="tracer_provider",
+            client_options=client_options,
+        )
+
+
+def test_resolve_feature_flags_exp_enabled_with_provider(monkeypatch):
+    """Verify that experimental feature is enabled if the experimental environment variable is enabled and a provider is provided."""
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+    client_options = {"tracer_provider": object()}
+
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=client_options,
+    )
+    assert result is True
+
+
+def test_resolve_feature_flags_exp_enabled_without_provider(monkeypatch):
+    """Verify that experimental feature is enabled if the experimental environment variable is enabled and NO provider is provided."""
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=None,
+    )
+    assert result is True
+
+
+def test_resolve_feature_flags_exp_disabled_without_provider(monkeypatch):
+    """Verify that experimental feature is disabled if the experimental environment variable is disabled and NO provider is provided."""
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
+
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=None,
+    )
+    assert result is False
+
+
+def test_resolve_feature_flags_ga_enabled_via_provider(monkeypatch):
+    """Verify that a GA feature is enabled if a provider is provided, ignoring the environment variable."""
+    # Env var is False, but provider is present
+    monkeypatch.setenv("GOOGLE_SDK_PYTHON_TRACING_ENABLED", "false")
+    client_options = {"tracer_provider": object()}
+
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=client_options,
+    )
+    assert result is True
+
+
+@pytest.mark.parametrize(
+    "env_val", [None, "false"], ids=["env_not_set", "env_explicit_false"]
+)
+def test_resolve_feature_flags_ga_fallback_to_false(monkeypatch, env_val):
+    """Verify that a GA feature is disabled if neither a provider is provided nor the environment variable is enabled."""
+    if env_val is not None:
+        monkeypatch.setenv("GOOGLE_SDK_PYTHON_TRACING_ENABLED", env_val)
+    else:
+        monkeypatch.delenv("GOOGLE_SDK_PYTHON_TRACING_ENABLED", raising=False)
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=None,
+    )
+    assert result is False
+
+
+class _MockOptions:
+    def __init__(self):
+        self.other_option = "value"
+
+
+@pytest.mark.parametrize(
+    "client_options",
+    [
+        {"other_option": "value"},
+        _MockOptions(),
+    ],
+    ids=["dict_without_key", "object_without_key"],
+)
+def test_resolve_feature_flags_options_without_key(client_options):
+    """Verify behavior when client_options is present but missing the provider key."""
+    # GA Path: should fall through to env var / fallback
+    result = feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        provider_key="tracer_provider",
+        client_options=client_options,
+    )
+    assert result is False