diff --git a/packages/prime/src/prime_cli/__init__.py b/packages/prime/src/prime_cli/__init__.py index e0d9d381d..70dabd2f6 100644 --- a/packages/prime/src/prime_cli/__init__.py +++ b/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.6.8" @@ -34,9 +39,12 @@ "CommandTimeoutError", "Config", "CreateSandboxRequest", + "FeatureFlagsClient", "Sandbox", "SandboxClient", "SandboxNotRunningError", "SandboxStatus", "UpdateSandboxRequest", + "evaluate_feature_flags", + "is_feature_enabled", ] diff --git a/packages/prime/src/prime_cli/feature_flags.py b/packages/prime/src/prime_cli/feature_flags.py new file mode 100644 index 000000000..14bad0d91 --- /dev/null +++ b/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 diff --git a/packages/prime/tests/test_feature_flags.py b/packages/prime/tests/test_feature_flags.py new file mode 100644 index 000000000..699312f4b --- /dev/null +++ b/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