feat: Add Feature Gating configuration helpers.

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

@@ -0,0 +1,139 @@
+# -*- 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
+
+# We import ClientOptions for type hinting
+from google.api_core.client_options import ClientOptions
+
+# Allowed truthy and falsy patterns for environment variables
+_TRUTHY_VALUES = ("y", "yes", "t", "true", "on", "1")
+_FALSY_VALUES = ("n", "no", "f", "false", "off", "0")
+
+
+class FeatureGatingError(ValueError):
+    """Raised when feature gating resolution fails or is misconfigured."""
+
+    pass
+
+
+def _strtobool(val: str) -> bool | None:
+    """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) -> bool | None:
+    """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_feature_key(
+    *, configuration: ClientOptions | dict[str, Any] | None, feature_key: str
+) -> bool:
+    """Checks if a specific feature key is present and not None in configuration."""
+    if configuration is None:
+        return False
+
+    if feature_key.startswith("__"):
+        return False
+
+    if isinstance(configuration, dict):
+        return configuration.get(feature_key) is not None
+
+    return getattr(configuration, feature_key, None) is not None
+
+
+def resolve_feature_flags(
+    *,
+    env_var: str,
+    feature_key: str,
+    configuration: ClientOptions | dict[str, Any] | None = None,
+) -> bool:
+    """Determines if a feature is enabled based on environment variables and configuration.
+
+    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 feature key is passed but the environment variable is not 'true',
+      raises FeatureGatingError (Fail Fast).
+
+    - **GA Path** (env_var does not contain "EXPERIMENTAL"):
+      Standard precedence. Enabled if a programmatic feature key is passed,
+      otherwise falls back to the environment variable value.
+
+    Args:
+        env_var: The name of the environment variable controlling this feature.
+        feature_key: The key in configuration/attributes for the programmatic configuration.
+        configuration: Optional. A dictionary or object containing client configuration.
+
+    Returns:
+        bool: True if the feature is resolved to enabled, False otherwise.
+
+    Raises:
+        FeatureGatingError: If a feature key is provided for an experimental feature without enabling the experimental environment variable.
+    """
+
+    # Check for programmatic feature configuration
+    has_feature_key = _has_feature_key(
+        configuration=configuration, feature_key=feature_key
+    )
+
+    # Read environment variable
+    env_var_setting = _get_env_bool(env_var)
+
+    # EXPERIMENTAL PATH:
+    # Resolution Hierarchy:
+    #   1. EXPERIMENTAL Environment Variable
+    #   2. Fail Fast if Feature Key present but EXPERIMENTAL Environment Variable is not enabled
+    if "EXPERIMENTAL" in env_var:
+        # Fail Fast if feature key present but experimental environment variable is not enabled
+        if env_var_setting is not True and has_feature_key:
+            raise FeatureGatingError(
+                f"Experimental feature requires {env_var} to be set to 'true' to use programmatic configuration."
+            )
+
+        return bool(env_var_setting)
+
+    # GENERAL AVAILABILITY PATH:
+    # Resolution Hierarchy:
+    #   1. Programmatic Configuration (Feature Key)
+    #   2. Environment Variable
+
+    # Check Programmatic Configuration
+    if has_feature_key:
+        return True
+
+    # Check Environment Variable
+    return bool(env_var_setting)

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

@@ -0,0 +1,224 @@
+# 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 (
+    FeatureGatingError,
+    _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",
+        feature_key="tracer_provider",
+        configuration=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_feature_key_fails_fast(
+    monkeypatch, exp_env_state
+):
+    """Verify that passing a feature_key to an experimental feature raises FeatureGatingError 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
+        )
+    configuration = {"tracer_provider": object()}
+
+    # Action & Assertion
+    with pytest.raises(FeatureGatingError, match="Experimental feature"):
+        _feature_gating_helpers.resolve_feature_flags(
+            env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+            feature_key="tracer_provider",
+            configuration=configuration,
+        )
+
+
+def test_resolve_feature_flags_exp_enabled_with_feature_key(monkeypatch):
+    """Verify that experimental feature is enabled if the experimental environment variable is enabled and a feature_key is provided."""
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+    configuration = {"tracer_provider": object()}
+
+    result = _feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+        feature_key="tracer_provider",
+        configuration=configuration,
+    )
+    assert result is True
+
+
+def test_resolve_feature_flags_exp_enabled_without_feature_key(monkeypatch):
+    """Verify that experimental feature is enabled if the experimental environment variable is enabled and NO feature_key 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",
+        feature_key="tracer_provider",
+        configuration=None,
+    )
+    assert result is True
+
+
+def test_resolve_feature_flags_exp_disabled_without_feature_key(monkeypatch):
+    """Verify that experimental feature is disabled if the experimental environment variable is disabled and NO feature_key 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",
+        feature_key="tracer_provider",
+        configuration=None,
+    )
+    assert result is False
+
+
+def test_resolve_feature_flags_ga_enabled_via_feature_key(monkeypatch):
+    """Verify that a GA feature is enabled if a feature_key is provided, ignoring the environment variable."""
+    # Env var is False, but feature_key is present in configuration
+    monkeypatch.setenv("GOOGLE_SDK_PYTHON_TRACING_ENABLED", "false")
+    configuration = {"tracer_provider": object()}
+
+    result = _feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        feature_key="tracer_provider",
+        configuration=configuration,
+    )
+    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 feature_key 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",
+        feature_key="tracer_provider",
+        configuration=None,
+    )
+    assert result is False
+
+
+class _MockOptions:
+    def __init__(self):
+        self.other_option = "value"
+
+
+@pytest.mark.parametrize(
+    "configuration",
+    [
+        {"other_option": "value"},
+        _MockOptions(),
+    ],
+    ids=["dict_without_key", "object_without_key"],
+)
+def test_resolve_feature_flags_options_without_key(configuration):
+    """Verify behavior when configuration is present but missing the feature key."""
+    # GA Path: should fall through to env var / fallback
+    result = _feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        feature_key="tracer_provider",
+        configuration=configuration,
+    )
+    assert result is False
+
+
+def test_resolve_feature_flags_rejects_dunder_keys(monkeypatch):
+    """Verify that dunder keys are rejected early in _has_feature_key."""
+    # We use a dunder key that exists on all objects (__class__)
+    # If the guardrail is missing, getattr might return it and return True
+    configuration = {"__class__": "some_value"}
+
+    result = _feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        feature_key="__class__",
+        configuration=configuration,
+    )
+    assert result is False
+
+    class MockWithClass:
+        __class__ = "some_value"
+
+    result = _feature_gating_helpers.resolve_feature_flags(
+        env_var="GOOGLE_SDK_PYTHON_TRACING_ENABLED",
+        feature_key="__class__",
+        configuration=MockWithClass(),
+    )
+    assert result is False