Add Prime feature flag evaluation client

Author: d42me

packages/prime/src/prime_cli/__init__.py

@@ -20,6 +20,11 @@
     AsyncAPIClient,
     Config,
 )
+from prime_cli.feature_flags import (
+    FeatureFlagsClient,
+    evaluate_feature_flags,
+    is_feature_enabled,
+)
 
 __version__ = "0.5.75"
 
@@ -34,9 +39,12 @@
     "CommandTimeoutError",
     "Config",
     "CreateSandboxRequest",
+    "FeatureFlagsClient",
     "Sandbox",
     "SandboxClient",
     "SandboxNotRunningError",
     "SandboxStatus",
     "UpdateSandboxRequest",
+    "evaluate_feature_flags",
+    "is_feature_enabled",
 ]

packages/prime/src/prime_cli/feature_flags.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from typing import TypeAlias
+
+from prime_cli.core import APIClient, APIError, Config
+
+JsonValue: TypeAlias = bool | int | float | str | None | list["JsonValue"] | dict[str, "JsonValue"]
+FeatureFlagDefaults: TypeAlias = dict[str, JsonValue]
+
+
+class FeatureFlagsClient:
+    """Authenticated client for Prime feature flag evaluation."""
+
+    def __init__(self, client: APIClient | None = None, config: Config | None = None) -> None:
+        self.client = client or APIClient()
+        self.config = config or self.client.config
+
+    def evaluate(self, defaults: FeatureFlagDefaults) -> FeatureFlagDefaults:
+        """Evaluate feature flags and fall back per key when the API omits a value."""
+        if not defaults:
+            return {}
+
+        from prime_cli import __version__
+
+        payload: dict[str, JsonValue] = {
+            "flags": defaults,
+            "cli_version": __version__,
+        }
+        if self.config.team_id:
+            payload["team_id"] = self.config.team_id
+
+        response = self.client.post("/feature-flags/evaluate", json=payload)
+        data = response.get("data")
+        if not isinstance(data, dict):
+            raise APIError("Feature flag response missing data")
+
+        flags = data.get("flags")
+        if not isinstance(flags, dict):
+            raise APIError("Feature flag response missing flags")
+
+        return {key: flags.get(key, default) for key, default in defaults.items()}
+
+
+def evaluate_feature_flags(
+    defaults: FeatureFlagDefaults,
+    client: APIClient | None = None,
+    config: Config | None = None,
+) -> FeatureFlagDefaults:
+    """Evaluate Prime feature flags, returning defaults if evaluation is unavailable."""
+    try:
+        return FeatureFlagsClient(client=client, config=config).evaluate(defaults)
+    except APIError:
+        return defaults.copy()
+
+
+def is_feature_enabled(
+    flag_key: str,
+    default: bool = False,
+    client: APIClient | None = None,
+    config: Config | None = None,
+) -> bool:
+    """Evaluate a boolean Prime feature flag."""
+    value = evaluate_feature_flags({flag_key: default}, client=client, config=config)[flag_key]
+    return value is True

packages/prime/tests/test_feature_flags.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from typing import Any
+
+from prime_cli import __version__
+from prime_cli.core import APIClient, APIError
+from prime_cli.feature_flags import (
+    FeatureFlagsClient,
+    evaluate_feature_flags,
+    is_feature_enabled,
+)
+
+
+class DummyConfig:
+    def __init__(self, team_id: str | None = None) -> None:
+        self.team_id = team_id
+
+
+class DummyFeatureFlagAPIClient(APIClient):
+    def __init__(
+        self,
+        response: dict[str, Any] | None = None,
+        team_id: str | None = None,
+        error: APIError | None = None,
+    ) -> None:
+        self.config = DummyConfig(team_id)
+        self.response = response or {"data": {"flags": {}}}
+        self.error = error
+        self.posts: list[tuple[str, dict[str, Any] | None]] = []
+
+    def post(self, endpoint: str, json: dict[str, Any] | None = None) -> dict[str, Any]:
+        self.posts.append((endpoint, json))
+        if self.error:
+            raise self.error
+        return self.response
+
+
+def test_feature_flags_client_evaluates_flags_with_cli_context() -> None:
+    client = DummyFeatureFlagAPIClient(
+        response={"data": {"flags": {"cli-new-flow": True}}},
+        team_id="team-1",
+    )
+
+    result = FeatureFlagsClient(client=client).evaluate(
+        {"cli-new-flow": False, "copy.variant": "control"}
+    )
+
+    assert result == {"cli-new-flow": True, "copy.variant": "control"}
+    assert client.posts == [
+        (
+            "/feature-flags/evaluate",
+            {
+                "flags": {"cli-new-flow": False, "copy.variant": "control"},
+                "cli_version": __version__,
+                "team_id": "team-1",
+            },
+        )
+    ]
+
+
+def test_evaluate_feature_flags_returns_defaults_when_api_unavailable() -> None:
+    client = DummyFeatureFlagAPIClient(error=APIError("feature flag service unavailable"))
+    defaults = {"cli-new-flow": False}
+
+    result = evaluate_feature_flags(defaults, client=client)
+
+    assert result == defaults
+    assert result is not defaults
+
+
+def test_feature_flags_client_skips_empty_request() -> None:
+    client = DummyFeatureFlagAPIClient()
+
+    assert FeatureFlagsClient(client=client).evaluate({}) == {}
+    assert client.posts == []
+
+
+def test_is_feature_enabled_only_accepts_boolean_true() -> None:
+    enabled_client = DummyFeatureFlagAPIClient(response={"data": {"flags": {"enabled": True}}})
+    string_client = DummyFeatureFlagAPIClient(response={"data": {"flags": {"enabled": "true"}}})
+
+    assert is_feature_enabled("enabled", client=enabled_client) is True
+    assert is_feature_enabled("enabled", default=True, client=string_client) is False