Support Datadog feauture flags

Author: KevinFairise2

src/dda/cli/application.py

@@ -16,6 +16,7 @@
 
     from dda.config.file import ConfigFile
     from dda.config.model import RootConfig
+    from dda.feature_flags.manager import FeatureFlagManager
     from dda.github.core import GitHub
     from dda.telemetry.manager import TelemetryManager
     from dda.tools import Tools
@@ -126,6 +127,12 @@ def telemetry(self) -> TelemetryManager:
 
         return TelemetryManager(self)
 
+    @cached_property
+    def ff(self) -> FeatureFlagManager:
+        from dda.feature_flags.manager import FeatureFlagManager
+
+        return FeatureFlagManager(self)
+
     @cached_property
     def dynamic_deps_allowed(self) -> bool:
         return os.getenv(AppEnvVars.NO_DYNAMIC_DEPS) not in {"1", "true"}

src/dda/config/constants.py

@@ -22,6 +22,7 @@ class AppEnvVars:
     VERBOSE = "DDA_VERBOSE"
     NO_DYNAMIC_DEPS = "DDA_NO_DYNAMIC_DEPS"
     TELEMETRY_API_KEY = "DDA_TELEMETRY_API_KEY"
+    FEATURE_FLAGS_CLIENT_TOKEN = "DDA_FEATURE_FLAGS_CLIENT_TOKEN"  # noqa: S105 This is not a hardcoded secret but the linter complains on it
     TELEMETRY_USER_MACHINE_ID = "DDA_TELEMETRY_USER_MACHINE_ID"
     # https://no-color.org
     NO_COLOR = "NO_COLOR"

src/dda/feature_flags/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT

src/dda/feature_flags/client.py

@@ -0,0 +1,126 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+import json
+from typing import Any, Optional
+
+from dda.utils.network.http.client import get_http_client
+
+
+class DatadogFeatureFlag:
+    """
+    Direct HTTP client for Datadog Feature Flag API
+
+    Based on the JavaScript implementation at:
+    /Users/kevin.fairise/dd/openfeature-js-client/packages/browser/src/transport/fetchConfiguration.ts
+    """
+
+    def __init__(
+        self,
+        client_token: str,
+    ):
+        """
+        Initialize the Datadog Feature Flag client
+
+        Args:
+            client_token: Your Datadog client token (starts with 'pub_')
+            site: Datadog site (e.g., 'datadoghq.com', 'datadoghq.eu')
+            env: Environment name
+            application_id: Your application ID for RUM attribution
+            service: Service name
+            version: Application version
+            flagging_proxy: Optional proxy URL for flagging configuration requests
+            custom_headers: Optional custom headers to add to requests
+        """
+        self.client_token = client_token
+        self.env = "Production"
+        self.endpoint_url = f"https://preview.ff-cdn.datadoghq.com/precompute-assignments?dd_env={self.env}"
+        self.application_id = "dda"
+        self.__client = get_http_client()
+
+    def _fetch_flags(
+        self, targeting_key: str = "", targeting_attributes: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        """
+        Fetch flag configuration from Datadog API
+
+        Args:
+            targeting_key: The targeting key (typically user ID)
+            targeting_attributes: Additional targeting attributes (context)
+
+        Returns:
+            Dictionary containing the flag configuration response
+
+        Raises:
+            requests.HTTPError: If the API request fails
+        """
+        # Build headers
+        headers = {
+            "Content-Type": "application/vnd.api+json",
+            "dd-client-token": self.client_token,
+        }
+
+        if self.application_id:
+            headers["dd-application-id"] = self.application_id
+
+        # Stringify all targeting attributes
+        stringified_attributes = {}
+        if targeting_attributes:
+            for key, value in targeting_attributes.items():
+                if isinstance(value, str):
+                    stringified_attributes[key] = value
+                else:
+                    stringified_attributes[key] = json.dumps(value)
+
+        # Build request payload (following JSON:API format)
+        payload = {
+            "data": {
+                "type": "precompute-assignments-request",
+                "attributes": {
+                    "env": {
+                        "dd_env": self.env,
+                    },
+                    "sdk": {
+                        "name": "python-example",
+                        "version": "0.1.0",
+                    },
+                    "subject": {
+                        "targeting_key": targeting_key,
+                        "targeting_attributes": stringified_attributes,
+                    },
+                },
+            },
+        }
+
+        try:
+            # Make the request
+            response = self.__client.post(self.endpoint_url, headers=headers, json=payload, timeout=10)
+        except Exception:  # noqa: BLE001
+            return {}
+
+        return response.json()
+
+    def get_flag_value(self, flag_key: str, context: dict[str, Any]) -> Any:
+        """
+        Get a flag value by key
+
+        Args:
+            flag_key: The flag key to evaluate
+            default_value: Default value if flag is not found
+            targeting_key: The targeting key (typically user ID)
+            targeting_attributes: Additional targeting attributes
+
+        Returns:
+            The flag value or default value
+        """
+        try:
+            response = self._fetch_flags(context["targeting_key"], context["targeting_attributes"])
+            # Navigate the response structure
+            flags = response.get("data", {}).get("attributes", {}).get("flags", {})
+            if flag_key in flags:
+                return flags[flag_key].get("variationValue", None)
+
+        except Exception:  # noqa: BLE001
+            return None
+
+        return None

src/dda/feature_flags/manager.py

@@ -0,0 +1,107 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any, Optional
+
+from dda.feature_flags.client import DatadogFeatureFlag
+from dda.utils.ci import running_in_ci
+from dda.utils.user import User
+
+if TYPE_CHECKING:
+    from dda.cli.application import Application
+    from dda.config.model import RootConfig
+
+
+class FeatureFlagUser(User):
+    def __init__(self, config: RootConfig) -> None:
+        super().__init__(config)
+
+
+class FeatureFlagManager:
+    """
+    A class for querying feature flags. This is available as the
+    [`Application.ff`][dda.cli.application.Application.ff] property.
+    """
+
+    def __init__(self, app: Application) -> None:
+        self.__app = app
+
+        self.__started = False
+
+        if self.client_token is not None:
+            self.__ff_client = DatadogFeatureFlag(self.client_token)
+            self.__started = True
+
+        self.__cache: dict[tuple[str, str, tuple[tuple[str, str],...]], Any] = {}
+
+    @cached_property
+    def client_token(self) -> str | None:
+        from contextlib import suppress
+
+        from dda.utils.secrets.secrets import fetch_client_token, read_client_token, save_client_token
+
+        client_token: str | None = None
+        with suppress(Exception):
+            client_token = read_client_token()
+            if not client_token:
+                client_token = fetch_client_token()
+                save_client_token(client_token)
+
+        return client_token
+
+    @property
+    def user(self) -> FeatureFlagUser:
+        return FeatureFlagUser(self.__app.config)
+
+    def get_targeting_key(self) -> str:
+        if running_in_ci():
+            import os
+
+            return os.getenv("CI_JOB_ID", "default_job_id")
+
+        return self.user.machine_id
+
+    def check_flag(self, flag_key: str, default_value: Any, extra_attributes: Optional[dict[str, str]] = None) -> bool:
+        if not self.__started:
+            return default_value
+
+        targeting_key = self.get_targeting_key()
+        targeting_attributes = self._get_base_context()
+        if extra_attributes is not None:
+            targeting_attributes.update(extra_attributes)
+
+        attributes_items = targeting_attributes.items()
+        tuple_attributes = tuple(((key, value) for key, value in sorted(attributes_items)))
+
+        self.__app.display_debug(
+            f"Checking flag {flag_key} with targeting key {targeting_key} and targeting attributes {tuple_attributes}"
+        )
+        flag_value = self._check_flag(flag_key, targeting_key, tuple_attributes)
+        if flag_value is None:
+            return default_value
+        return flag_value
+
+    def _check_flag(self, flag_key: str, targeting_key: str, targeting_attributes: tuple[tuple[str, str],...]) -> bool:
+        cache_key = (flag_key, targeting_key, targeting_attributes)
+        if cache_key in self.__cache:
+            return self.__cache[cache_key]
+
+        context = {
+            "targeting_key": targeting_key,
+            "targeting_attributes": dict(targeting_attributes),
+        }
+
+        flag_value = self.__ff_client.get_flag_value(flag_key, context)
+        self.__cache[cache_key] = flag_value
+        return flag_value
+
+    def _get_base_context(self) -> dict[str, str]:
+        return {
+            "platform": "toto",
+            "ci": "true" if running_in_ci() else "false",
+            "env": "prod",
+            "user": self.user.email,
+        }

src/dda/telemetry/daemon/__main__.py

@@ -16,8 +16,8 @@
 
 from dda.telemetry.constants import DaemonEnvVars
 from dda.telemetry.daemon.handler import finalize_error
-from dda.telemetry.secrets import fetch_api_key, read_api_key, save_api_key
 from dda.utils.fs import Path
+from dda.utils.secrets.secrets import fetch_api_key, read_api_key, save_api_key
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator

src/dda/telemetry/manager.py

@@ -7,6 +7,8 @@
 from functools import cached_property
 from typing import TYPE_CHECKING
 
+from dda.utils.user import User
+
 if TYPE_CHECKING:
     from dda.cli.application import Application
     from dda.config.model import RootConfig
@@ -76,7 +78,7 @@ def api_key(self) -> str | None:
 
         from contextlib import suppress
 
-        from dda.telemetry.secrets import fetch_api_key, read_api_key, save_api_key
+        from dda.utils.secrets.secrets import fetch_api_key, read_api_key, save_api_key
 
         api_key: str | None = None
         with suppress(Exception):
@@ -116,7 +118,6 @@ def __write_dir(self) -> Path:
         return Path(mkdtemp(prefix="dda-telemetry-"))
 
     def __start_daemon(self) -> None:
-        import os
         import sys
 
         from dda.telemetry.constants import DaemonEnvVars
@@ -132,25 +133,6 @@ def __start_daemon(self) -> None:
         self.__started = True
 
 
-class TelemetryUser:
+class TelemetryUser(User):
     def __init__(self, config: RootConfig) -> None:
-        self.__config = config
-
-    @cached_property
-    def machine_id(self) -> str:
-        from dda.config.constants import AppEnvVars
-
-        if machine_id := os.environ.get(AppEnvVars.TELEMETRY_USER_MACHINE_ID):
-            return machine_id
-
-        from dda.utils.platform import get_machine_id
-
-        return str(get_machine_id())
-
-    @cached_property
-    def name(self) -> str:
-        return self.__config.user.name if self.__config.user.name != "auto" else self.__config.tools.git.author.name
-
-    @cached_property
-    def email(self) -> str:
-        return self.__config.user.email if self.__config.user.email != "auto" else self.__config.tools.git.author.email
+        super().__init__(config)

src/dda/telemetry/secrets.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT