Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .sampo/changesets/feature-flag-evaluation-runtime.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
pypi/posthog: minor
---

Read a feature flag's evaluation runtime from the SDK: `get_feature_flag_evaluation_runtime(key)` returns the `FeatureFlagEvaluationRuntime` on a locally loaded flag definition, and `get_feature_flag_keys_by_evaluation_runtime(runtime)` lists the keys a given runtime can evaluate. The local-evaluation payload already carried the value, but only on the untyped `client.feature_flags` dicts, so a backend that serves flags to its own frontend had to call `/api/feature_flag/local_evaluation` itself to see it.
49 changes: 49 additions & 0 deletions posthog/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@
from posthog.types import (
BeforeSendCallback as BeforeSendCallback,
FeatureFlag as FeatureFlag,
FeatureFlagEvaluationRuntime as FeatureFlagEvaluationRuntime,
FlagValue as FlagValue,
FlagsAndPayloads as FlagsAndPayloads,
)
Expand Down Expand Up @@ -1157,6 +1158,54 @@ def feature_flag_definitions():
return _proxy("feature_flag_definitions")


def get_feature_flag_evaluation_runtime(
key: str,
) -> Optional[FeatureFlagEvaluationRuntime]:
"""
Return where a locally loaded feature flag is meant to be evaluated.

Details:
Reads the `evaluation_runtime` each flag definition carries, so no extra
request is made. Returns `None` when local evaluation has not loaded a
definition for this key. A definition that carries no runtime reports
`FeatureFlagEvaluationRuntime.ALL`, the default PostHog applies.

Examples:
```python
from posthog import FeatureFlagEvaluationRuntime, get_feature_flag_evaluation_runtime
runtime = get_feature_flag_evaluation_runtime("my-flag")
```

Category:
Feature flags
"""
return _proxy("get_feature_flag_evaluation_runtime", key)


def get_feature_flag_keys_by_evaluation_runtime(
evaluation_runtime: Union[FeatureFlagEvaluationRuntime, str],
) -> list[str]:
"""
Return the keys of locally loaded flags that a runtime can evaluate.

Details:
A flag set to `FeatureFlagEvaluationRuntime.ALL` suits either runtime, so
it is returned for `CLIENT` and for `SERVER`. Use this to decide which
flags to hand to a browser when a backend serves flags to its own
frontend.

Examples:
```python
from posthog import FeatureFlagEvaluationRuntime, get_feature_flag_keys_by_evaluation_runtime
client_keys = get_feature_flag_keys_by_evaluation_runtime(FeatureFlagEvaluationRuntime.CLIENT)
```

Category:
Feature flags
"""
return _proxy("get_feature_flag_keys_by_evaluation_runtime", evaluation_runtime)


def load_feature_flags():
"""
Load feature flag definitions from PostHog.
Expand Down
75 changes: 75 additions & 0 deletions posthog/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@
from posthog.types import (
FeatureFlag,
FeatureFlagError,
FeatureFlagEvaluationRuntime,
FeatureFlagResult,
FlagMetadata,
FlagsAndPayloads,
Expand Down Expand Up @@ -4704,6 +4705,80 @@ def feature_flag_definitions(self):
"""
return self.feature_flags

def get_feature_flag_evaluation_runtime(
self, key: str
) -> Optional[FeatureFlagEvaluationRuntime]:
"""
Return where a locally loaded feature flag is meant to be evaluated.

Args:
key: The feature flag key.

Returns:
The flag's evaluation runtime, or ``None`` when local evaluation has
not loaded a definition for this key. A definition that carries no
runtime reports ``FeatureFlagEvaluationRuntime.ALL``, the default
PostHog applies.

Examples:
```python
from posthog import FeatureFlagEvaluationRuntime

runtime = posthog.get_feature_flag_evaluation_runtime("my-flag")
if runtime is FeatureFlagEvaluationRuntime.SERVER:
...
```

Category:
Feature flags
"""
definition = (self.feature_flags_by_key or {}).get(key)
if definition is None:
return None
return FeatureFlagEvaluationRuntime.from_value(
definition.get("evaluation_runtime")
)

def get_feature_flag_keys_by_evaluation_runtime(
self, evaluation_runtime: Union[FeatureFlagEvaluationRuntime, str]
) -> list[str]:
"""
Return the keys of locally loaded flags that a runtime can evaluate.

A flag set to ``FeatureFlagEvaluationRuntime.ALL`` suits either runtime,
so it is returned for ``CLIENT`` and for ``SERVER``, and asking for
``ALL`` returns every loaded flag. Use this to decide which flags to hand
to a browser when a backend serves flags to its own frontend.

Args:
evaluation_runtime: The runtime to match, as a
``FeatureFlagEvaluationRuntime`` or its string value.

Returns:
The matching flag keys, in the order local evaluation loaded them.
Empty when no definitions are loaded.

Examples:
```python
from posthog import FeatureFlagEvaluationRuntime

client_keys = posthog.get_feature_flag_keys_by_evaluation_runtime(
FeatureFlagEvaluationRuntime.CLIENT
)
```

Category:
Feature flags
"""
wanted = FeatureFlagEvaluationRuntime(evaluation_runtime)
return [
key
for key, definition in (self.feature_flags_by_key or {}).items()
if FeatureFlagEvaluationRuntime.from_value(
definition.get("evaluation_runtime")
).matches(wanted)
]

def _person_properties_for_local_evaluation(self, distinct_id, person_properties):
local_person_properties = dict(person_properties or {})
local_person_properties.setdefault("distinct_id", distinct_id)
Expand Down
111 changes: 111 additions & 0 deletions posthog/test/test_feature_flags.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
relative_date_parse_for_feature_flag_matching,
)
from posthog.request import APIError, GetResponse
from posthog.types import FeatureFlagEvaluationRuntime
from posthog.test.test_utils import FAKE_TEST_API_KEY
from posthog.utils import FlagCache

Expand Down Expand Up @@ -5052,6 +5053,116 @@ def test_get_all_flags_fallback_when_device_id_missing_for_some_flags(
self.assertEqual(patch_flags.call_count, 1)


class TestEvaluationRuntime(unittest.TestCase):
def setUp(self):
self.client = Client(FAKE_TEST_API_KEY)
self.client.feature_flags = [
{"id": 1, "key": "everywhere-flag", "evaluation_runtime": "all"},
{"id": 2, "key": "browser-flag", "evaluation_runtime": "client"},
{"id": 3, "key": "backend-flag", "evaluation_runtime": "server"},
{"id": 4, "key": "unset-flag", "evaluation_runtime": None},
{"id": 5, "key": "legacy-flag"},
]

@parameterized.expand(
[
("everywhere-flag", FeatureFlagEvaluationRuntime.ALL),
("browser-flag", FeatureFlagEvaluationRuntime.CLIENT),
("backend-flag", FeatureFlagEvaluationRuntime.SERVER),
# A definition without a runtime reports the default PostHog applies.
("unset-flag", FeatureFlagEvaluationRuntime.ALL),
("legacy-flag", FeatureFlagEvaluationRuntime.ALL),
]
)
def test_runtime_of_loaded_flag(self, key, expected):
self.assertEqual(self.client.get_feature_flag_evaluation_runtime(key), expected)

def test_runtime_of_unknown_flag_is_none(self):
self.assertIsNone(
self.client.get_feature_flag_evaluation_runtime("no-such-flag")
)

def test_runtime_before_definitions_are_loaded_is_none(self):
client = Client(FAKE_TEST_API_KEY)
self.assertIsNone(client.get_feature_flag_evaluation_runtime("browser-flag"))
self.assertEqual(
client.get_feature_flag_keys_by_evaluation_runtime("client"), []
)

@parameterized.expand(
[
(
FeatureFlagEvaluationRuntime.CLIENT,
["everywhere-flag", "browser-flag", "unset-flag", "legacy-flag"],
),
(
FeatureFlagEvaluationRuntime.SERVER,
["everywhere-flag", "backend-flag", "unset-flag", "legacy-flag"],
),
(
FeatureFlagEvaluationRuntime.ALL,
[
"everywhere-flag",
"browser-flag",
"backend-flag",
"unset-flag",
"legacy-flag",
],
),
(
"client",
["everywhere-flag", "browser-flag", "unset-flag", "legacy-flag"],
),
]
)
def test_keys_by_evaluation_runtime(self, runtime, expected):
self.assertEqual(
self.client.get_feature_flag_keys_by_evaluation_runtime(runtime), expected
)

def test_unrecognized_runtime_argument_raises(self):
with self.assertRaises(ValueError):
self.client.get_feature_flag_keys_by_evaluation_runtime("serverless")

@mock.patch("posthog.client.Poller")
@mock.patch("posthog.client.get")
def test_runtime_survives_a_local_evaluation_fetch(self, patch_get, patch_poll):
patch_get.return_value = GetResponse(
data={
"flags": [
{
"id": 1,
"name": "Browser Feature",
"key": "browser-feature",
"active": True,
"evaluation_runtime": "client",
}
],
"group_type_mapping": {},
"cohorts": {},
}
)
client = Client(FAKE_TEST_API_KEY, secret_key="test")
client.load_feature_flags()

self.assertEqual(
client.get_feature_flag_evaluation_runtime("browser-feature"),
FeatureFlagEvaluationRuntime.CLIENT,
)
self.assertEqual(
client.get_feature_flag_keys_by_evaluation_runtime(
FeatureFlagEvaluationRuntime.CLIENT
),
["browser-feature"],
)
self.assertEqual(
client.get_feature_flag_keys_by_evaluation_runtime(
FeatureFlagEvaluationRuntime.SERVER
),
[],
)


class TestMatchProperties(unittest.TestCase):
def property(self, key, value, operator=None):
result = {"key": key, "value": value}
Expand Down
36 changes: 36 additions & 0 deletions posthog/types.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import json
from dataclasses import dataclass
from enum import Enum
from typing import Any, Callable, List, Optional, TypedDict, Union, cast

FlagValue = Union[bool, str]
Expand Down Expand Up @@ -42,6 +43,41 @@ class SendFeatureFlagsOptions(TypedDict, total=False):
flag_keys_filter: Optional[list[str]]


class FeatureFlagEvaluationRuntime(str, Enum):
"""Where a feature flag is meant to be evaluated.

Set per flag in PostHog and carried on every locally cached flag definition.
``ALL`` means the flag suits both client-side and server-side evaluation, so
it matches either runtime. Inheriting from ``str`` keeps the members directly
comparable to their ``"all"`` / ``"client"`` / ``"server"`` values.
"""

ALL = "all"
CLIENT = "client"
SERVER = "server"

@classmethod
def from_value(cls, value: Any) -> "FeatureFlagEvaluationRuntime":
"""Coerce a raw ``evaluation_runtime`` value to a member.

A missing, null or unrecognized value becomes ``ALL``, which is the
default PostHog applies to a flag that does not set a runtime.
"""
if isinstance(value, str):
try:
return cls(value.strip().lower())
except ValueError:
pass
return cls.ALL

def matches(self, other: "FeatureFlagEvaluationRuntime") -> bool:
"""Whether a flag set to one of these runtimes suits the other runtime.

``ALL`` matches every runtime, so the check is symmetric.
"""
return self is other or FeatureFlagEvaluationRuntime.ALL in (self, other)


@dataclass(frozen=True)
class FlagReason:
"""Reason metadata returned by the feature flag API.
Expand Down
Loading
Loading