feat(hc): Adds Flagpole core data model (#68551)

Author: GabeVillalobos

src/flagpole/__init__.py

@@ -0,0 +1,124 @@
+"""
+Options backed feature flagging.
+
+Entry backed options. Will consume option data that is structured like
+
+```yaml
+features:
+  organizations:fury-mode:
+    enabled: True
+    name: sentry organizations
+    owner: hybrid-cloud
+    segments:
+      - name: sentry orgs
+        rollout: 50
+        conditions:
+          - property: organization_slug
+            name: internal organizations
+            operator:
+                kind: in
+                value: ["sentry-test", "sentry"]
+      - name: free accounts
+        conditions:
+          - property: subscription_is_free
+            name: free subscriptions
+            operator:
+              kind: equals
+              value: True
+```
+
+Each feature flag has a list of segments, each of which contain a list of conditions.
+If all the conditions for a segment match the evaluation context, a feature is granted.
+A segment with multiple conditions looks like:
+
+```yaml
+features:
+  organizations:fury-mode:
+    enabled: True
+    owner: hybrid-cloud
+    description: sentry organizations
+    segments:
+      - name: sentry organizations
+        rollout: 50
+        conditions:
+          - name: internal orgs
+            property: organization_slug
+            operator:
+              kind: in
+              value: ["sentry-test", "sentry"]
+          - name: allowed users
+            property: user_email
+            operator:
+              kind: in
+              value: ["mark@sentry.io", "gabe@sentry.io"]
+```
+
+Property names are arbitrary and read from an evaluation context
+prepared by the application.
+
+Each condition has a single operator. An operator takes a kind (`OperatorKind` enum)
+and a value, the type of which depends on the operator specified.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, Field, ValidationError, constr
+
+from flagpole.conditions import Segment
+from flagpole.evaluation_context import ContextBuilder, EvaluationContext
+from sentry.utils import json
+
+
+class InvalidFeatureFlagConfiguration(Exception):
+    pass
+
+
+class Feature(BaseModel):
+    name: constr(min_length=1, to_lower=True)  # type:ignore[valid-type]
+    owner: constr(min_length=1)  # type:ignore[valid-type]
+    segments: list[Segment]
+    """A list of segments to evaluate against the provided data"""
+    enabled: bool = True
+    """Defines whether or not the feature is enabled."""
+    created_at: datetime = Field(default_factory=datetime.now)
+    """This datetime is when this instance was created. It can be used to decide when to re-read configuration data"""
+
+    def match(self, context: EvaluationContext) -> bool:
+        if self.enabled:
+            for segment in self.segments:
+                if segment.match(context):
+                    return True
+
+        return False
+
+    def dump_schema_to_file(self, file_path: str) -> None:
+        with open(file_path, "w") as file:
+            file.write(self.schema_json())
+
+    @classmethod
+    def from_feature_dictionary(cls, name: str, config_dict: dict[str, Any]) -> Feature:
+        try:
+            feature = cls(name=name, **config_dict)
+        except ValidationError as exc:
+            raise InvalidFeatureFlagConfiguration("Provided JSON is not a valid feature") from exc
+
+        return feature
+
+    @classmethod
+    def from_feature_config_json(
+        cls, name: str, config_json: str, context_builder: ContextBuilder | None = None
+    ) -> Feature:
+        try:
+            config_data_dict = json.loads(config_json)
+        except json.JSONDecodeError as decode_error:
+            raise InvalidFeatureFlagConfiguration("Invalid feature json provided") from decode_error
+
+        if not isinstance(config_data_dict, dict):
+            raise InvalidFeatureFlagConfiguration("Feature JSON is not a valid feature")
+
+        return cls.from_feature_dictionary(name=name, config_dict=config_data_dict)
+
+
+__all__ = ["Feature", "InvalidFeatureFlagConfiguration", "ContextBuilder", "EvaluationContext"]

src/flagpole/conditions.py

@@ -0,0 +1,34 @@
+from pydantic import BaseModel, constr
+
+from flagpole.evaluation_context import EvaluationContext
+from flagpole.operators import AvailableOperators
+
+
+class Condition(BaseModel):
+    property: str | None
+    operator: AvailableOperators
+
+    def match(self, context: EvaluationContext, segment_name: str) -> bool:
+        if self.property is None:
+            return False
+
+        return self.operator.match(
+            condition_property=context.get(self.property), segment_name=segment_name
+        )
+
+
+class Segment(BaseModel):
+    name: constr(min_length=1)  # type:ignore[valid-type]
+    conditions: list[Condition]
+    rollout: int | None = 0
+
+    def match(self, context: EvaluationContext) -> bool:
+        for condition in self.conditions:
+            match_condition = condition.match(context, segment_name=self.name)
+            if not match_condition:
+                return False
+        # Apply incremental rollout if available.
+        if self.rollout is not None and self.rollout < 100:
+            return context.id() % 100 <= self.rollout
+
+        return True

src/flagpole/evaluation_context.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Callable
+from copy import deepcopy
+from typing import Any, TypeVar
+
+from pydantic import BaseModel
+
+ValidContextTypes = TypeVar(
+    "ValidContextTypes",
+    bound=str | int | float | bool | list[str] | list[int] | list[float] | list[bool],
+)
+
+EvaluationContextDict = dict[str, ValidContextTypes]
+
+
+class EvaluationContext:
+    """
+    Prepared by the application and passed to flagpole to evaluate
+    feature conditions.
+    """
+
+    def __init__(self, data: EvaluationContextDict):
+        self.__data = deepcopy(data)
+
+    def get(self, key: str) -> Any:
+        return self.__data.get(key)
+
+    def has(self, key: str) -> Any:
+        return key in self.__data
+
+    def size(self) -> int:
+        return len(self.__data)
+
+    def id(self) -> int:
+        """
+        Return a hashed identifier for this context
+
+        The identifier should be stable for a given context contents.
+        Identifiers are used to determine rollout groups deterministically
+        and consistently.
+        """
+        keys = self.__data.keys()
+        vector = []
+        for key in sorted(keys):
+            vector.append(key)
+            vector.append(str(self.__data[key]))
+        hashed = hashlib.sha1(":".join(vector).encode("utf8"))
+        return int.from_bytes(hashed.digest(), byteorder="big")
+
+
+# A function that generates a new slice of evaluation context data as a dictionary.
+EvaluationContextTransformer = Callable[[dict[str, Any]], EvaluationContextDict]
+
+
+class ContextBuilder(BaseModel):
+    """
+    Used to build an EvaluationContext instance for use in Flagpole.
+    This class aggregates a list of context transformers, each of which are
+    responsible for generating a slice of context data.
+
+    This class is meant to be used with Flagpole's `Feature` class:
+    >>> from flagpole import ContextBuilder, Feature
+    >>> builder = ContextBuilder().add_context_transformer(lambda _dict: dict(foo="bar"))
+    >>> feature = Feature.from_feature_dictionary(name="foo", feature_dictionary=dict(), context=builder)
+    >>> feature.match(dict())
+    """
+
+    context_transformers: list[EvaluationContextTransformer] = []
+    exception_handler: Callable[[Exception], Any] | None
+
+    def add_context_transformer(
+        self, context_transformer: EvaluationContextTransformer
+    ) -> ContextBuilder:
+        self.context_transformers.append(context_transformer)
+        return self
+
+    def add_exception_handler(self, exception_handler: Callable[[Exception], None]):
+        """
+        Add a custom exception handler to the context builder if you need custom handling
+        if any of the transformer functions raise an exception. This is useful for swallowing
+        or reporting any exceptions that occur while building a context.
+
+        :param exception_handler:
+        """
+        if self.exception_handler is not None:
+            raise Exception("Exception handler is already defined")
+
+        self.exception_handler = exception_handler
+
+    def build(self, data: dict[str, Any] | None = None) -> EvaluationContext:
+        builder_data: dict[str, Any] = data or dict()
+        context_data: dict[str, Any] = dict()
+
+        for transformer in self.context_transformers:
+            try:
+                context_data = {**context_data, **transformer(builder_data)}
+            except Exception as e:
+                if self.exception_handler is not None:
+                    self.exception_handler(e)
+                else:
+                    raise
+
+        return EvaluationContext(context_data)