feat: support-evaluation-tracking-api (#196)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: Zaimwa9

flagsmith/__init__.py

@@ -1,5 +1,6 @@
 from flagsmith import webhooks
+from flagsmith.analytics import PipelineAnalyticsConfig
 from flagsmith.flagsmith import Flagsmith
 from flagsmith.version import __version__
 
-__all__ = ("Flagsmith", "webhooks", "__version__")
+__all__ = ("Flagsmith", "PipelineAnalyticsConfig", "webhooks", "__version__")

flagsmith/analytics.py

@@ -1,9 +1,18 @@
+import atexit
 import json
+import logging
+import threading
+import time
 import typing
+from dataclasses import dataclass
 from datetime import datetime
 
 from requests_futures.sessions import FuturesSession  # type: ignore
 
+from flagsmith.version import __version__
+
+logger = logging.getLogger(__name__)
+
 ANALYTICS_ENDPOINT: typing.Final[str] = "analytics/flags/"
 
 # Used to control how often we send data(in seconds)
@@ -60,3 +69,161 @@ def track_feature(self, feature_name: str) -> None:
         self.analytics_data[feature_name] = self.analytics_data.get(feature_name, 0) + 1
         if (datetime.now() - self._last_flushed).seconds > ANALYTICS_TIMER:
             self.flush()
+
+
+@dataclass
+class PipelineAnalyticsConfig:
+    analytics_server_url: str
+    max_buffer_items: int = 1000
+    flush_interval_seconds: float = 10.0
+
+
+class PipelineAnalyticsProcessor:
+    """
+    Buffered analytics processor that sends per-evaluation and custom events
+    to the Flagsmith pipeline analytics endpoint in batches.
+
+    Evaluation events are deduplicated within each flush window. Events are
+    flushed periodically via a background timer or when the buffer is full.
+    """
+
+    def __init__(
+        self,
+        config: PipelineAnalyticsConfig,
+        environment_key: str,
+    ) -> None:
+        url = config.analytics_server_url
+        if not url.endswith("/"):
+            url = f"{url}/"
+        self._batch_endpoint = f"{url}v1/analytics/batch"
+        self._environment_key = environment_key
+        self._max_buffer = config.max_buffer_items
+        self._flush_interval_seconds = config.flush_interval_seconds
+
+        self._buffer: typing.List[typing.Dict[str, typing.Any]] = []
+        self._dedup_keys: typing.Dict[str, str] = {}
+        self._lock = threading.Lock()
+        self._timer: typing.Optional[threading.Timer] = None
+
+    def record_evaluation_event(
+        self,
+        flag_key: str,
+        enabled: bool,
+        value: typing.Any,
+        identity_identifier: typing.Optional[str] = None,
+        traits: typing.Optional[typing.Dict[str, typing.Any]] = None,
+    ) -> None:
+        fingerprint = f"{identity_identifier or 'none'}|{enabled}|{value}"
+        should_flush = False
+
+        with self._lock:
+            if self._dedup_keys.get(flag_key) == fingerprint:
+                return
+            self._dedup_keys[flag_key] = fingerprint
+            self._buffer.append(
+                {
+                    "event_id": flag_key,
+                    "event_type": "flag_evaluation",
+                    "evaluated_at": int(time.time() * 1000),
+                    "identity_identifier": identity_identifier,
+                    "enabled": enabled,
+                    "value": value,
+                    "traits": dict(traits) if traits else None,
+                    "metadata": {"sdk_version": __version__},
+                }
+            )
+            if len(self._buffer) >= self._max_buffer:
+                should_flush = True
+
+        if should_flush:
+            self.flush()
+
+    def record_custom_event(
+        self,
+        event_name: str,
+        identity_identifier: typing.Optional[str] = None,
+        traits: typing.Optional[typing.Dict[str, typing.Any]] = None,
+        metadata: typing.Optional[typing.Dict[str, typing.Any]] = None,
+    ) -> None:
+        should_flush = False
+
+        with self._lock:
+            self._buffer.append(
+                {
+                    "event_id": event_name,
+                    "event_type": "custom_event",
+                    "evaluated_at": int(time.time() * 1000),
+                    "identity_identifier": identity_identifier,
+                    "enabled": None,
+                    "value": None,
+                    "traits": dict(traits) if traits else None,
+                    "metadata": {**(metadata or {}), "sdk_version": __version__},
+                }
+            )
+            if len(self._buffer) >= self._max_buffer:
+                should_flush = True
+
+        if should_flush:
+            self.flush()
+
+    def flush(self) -> None:
+        with self._lock:
+            if not self._buffer:
+                return
+            events = self._buffer
+            self._buffer = []
+            self._dedup_keys.clear()
+
+        payload = json.dumps(
+            {"events": events, "environment_key": self._environment_key}
+        )
+        try:
+            future = session.post(
+                self._batch_endpoint,
+                data=payload,
+                timeout=3,
+                headers={
+                    "Content-Type": "application/json; charset=utf-8",
+                    "X-Environment-Key": self._environment_key,
+                    "Flagsmith-SDK-User-Agent": f"flagsmith-python-client/{__version__}",
+                },
+            )
+        except RuntimeError:
+            logger.debug("Skipping flush: thread pool already shut down")
+            return
+        future.add_done_callback(lambda f: self._handle_flush_result(f, events))
+
+    def _handle_flush_result(
+        self,
+        future: typing.Any,
+        events: typing.List[typing.Dict[str, typing.Any]],
+    ) -> None:
+        try:
+            response = future.result()
+            response.raise_for_status()
+        except Exception:
+            logger.warning(
+                "Failed to flush pipeline analytics, re-queuing events", exc_info=True
+            )
+            with self._lock:
+                self._buffer = events + self._buffer
+                self._buffer = self._buffer[: self._max_buffer]
+
+    def start(self) -> None:
+        self._schedule_flush()
+        atexit.register(self.stop)
+
+    def stop(self) -> None:
+        atexit.unregister(self.stop)
+        if self._timer is not None:
+            self._timer.cancel()
+        self.flush()
+
+    def _schedule_flush(self) -> None:
+        self._timer = threading.Timer(self._flush_interval_seconds, self._timer_flush)
+        self._timer.daemon = True
+        self._timer.start()
+
+    def _timer_flush(self) -> None:
+        self.flush()
+        self._schedule_flush()

flagsmith/flagsmith.py

@@ -8,13 +8,18 @@
 from requests.adapters import HTTPAdapter
 from urllib3 import Retry
 
-from flagsmith.analytics import AnalyticsProcessor
+from flagsmith.analytics import (
+    AnalyticsProcessor,
+    PipelineAnalyticsConfig,
+    PipelineAnalyticsProcessor,
+)
 from flagsmith.exceptions import FlagsmithAPIError, FlagsmithClientError
 from flagsmith.mappers import (
     map_context_and_identity_data_to_context,
     map_environment_document_to_context,
     map_environment_document_to_environment_updated_at,
     map_segment_results_to_identity_segments,
+    resolve_trait_values,
 )
 from flagsmith.models import DefaultFlag, Flags, Segment
 from flagsmith.offline_handlers import OfflineHandler
@@ -63,6 +68,7 @@ def __init__(
         environment_refresh_interval_seconds: typing.Union[int, float] = 60,
         retries: typing.Optional[Retry] = None,
         enable_analytics: bool = False,
+        pipeline_analytics_config: typing.Optional[PipelineAnalyticsConfig] = None,
         default_flag_handler: typing.Optional[
             typing.Callable[[str], DefaultFlag]
         ] = None,
@@ -108,6 +114,9 @@ def __init__(
         self.default_flag_handler = default_flag_handler
         self.enable_realtime_updates = enable_realtime_updates
         self._analytics_processor: typing.Optional[AnalyticsProcessor] = None
+        self._pipeline_analytics_processor: typing.Optional[
+            PipelineAnalyticsProcessor
+        ] = None
         self._evaluation_context: typing.Optional[SDKEvaluationContext] = None
         self._environment_updated_at: typing.Optional[datetime] = None
 
@@ -170,10 +179,28 @@ def __init__(
 
                 self._initialise_local_evaluation()
 
-            if enable_analytics:
-                self._analytics_processor = AnalyticsProcessor(
-                    environment_key, self.api_url, timeout=self.request_timeout_seconds
-                )
+            self._initialise_analytics(
+                environment_key=environment_key,
+                enable_analytics=enable_analytics,
+                pipeline_analytics_config=pipeline_analytics_config,
+            )
+
+    def _initialise_analytics(
+        self,
+        environment_key: str,
+        enable_analytics: bool,
+        pipeline_analytics_config: typing.Optional[PipelineAnalyticsConfig],
+    ) -> None:
+        if enable_analytics:
+            self._analytics_processor = AnalyticsProcessor(
+                environment_key, self.api_url, timeout=self.request_timeout_seconds
+            )
+        if pipeline_analytics_config:
+            self._pipeline_analytics_processor = PipelineAnalyticsProcessor(
+                config=pipeline_analytics_config,
+                environment_key=environment_key,
+            )
+            self._pipeline_analytics_processor.start()
 
     def _initialise_local_evaluation(self) -> None:
         # To ensure that the environment is set before allowing subsequent
@@ -290,6 +317,25 @@ def get_identity_segments(
 
         return map_segment_results_to_identity_segments(evaluation_result["segments"])
 
+    def track_event(
+        self,
+        event_name: str,
+        identity_identifier: typing.Optional[str] = None,
+        traits: typing.Optional[TraitMapping] = None,
+        metadata: typing.Optional[typing.Dict[str, typing.Any]] = None,
+    ) -> None:
+        if not self._pipeline_analytics_processor:
+            raise ValueError(
+                "Pipeline analytics is not configured. "
+                "Provide pipeline_analytics_config to use track_event."
+            )
+        self._pipeline_analytics_processor.record_custom_event(
+            event_name=event_name,
+            identity_identifier=identity_identifier,
+            traits=resolve_trait_values(traits),
+            metadata=metadata,
+        )
+
     def update_environment(self) -> None:
         try:
             environment_data = self._get_json_response(
@@ -345,6 +391,7 @@ def _get_environment_flags_from_document(self) -> Flags:
             evaluation_result=evaluation_result,
             analytics_processor=self._analytics_processor,
             default_flag_handler=self.default_flag_handler,
+            pipeline_analytics_processor=self._pipeline_analytics_processor,
         )
 
     def _get_identity_flags_from_document(
@@ -368,6 +415,9 @@ def _get_identity_flags_from_document(
             evaluation_result=evaluation_result,
             analytics_processor=self._analytics_processor,
             default_flag_handler=self.default_flag_handler,
+            pipeline_analytics_processor=self._pipeline_analytics_processor,
+            identity_identifier=identifier,
+            traits=resolve_trait_values(traits),
         )
 
     def _get_environment_flags_from_api(self) -> Flags:
@@ -379,6 +429,7 @@ def _get_environment_flags_from_api(self) -> Flags:
                 api_flags=json_response,
                 analytics_processor=self._analytics_processor,
                 default_flag_handler=self.default_flag_handler,
+                pipeline_analytics_processor=self._pipeline_analytics_processor,
             )
         except FlagsmithAPIError:
             if self.offline_handler:
@@ -411,6 +462,9 @@ def _get_identity_flags_from_api(
                 api_flags=json_response["flags"],
                 analytics_processor=self._analytics_processor,
                 default_flag_handler=self.default_flag_handler,
+                pipeline_analytics_processor=self._pipeline_analytics_processor,
+                identity_identifier=identifier,
+                traits=resolve_trait_values(traits),
             )
         except FlagsmithAPIError:
             if self.offline_handler:
@@ -443,3 +497,6 @@ def __del__(self) -> None:
 
         if hasattr(self, "event_stream_thread"):
             self.event_stream_thread.stop()
+
+        if self._pipeline_analytics_processor:
+            self._pipeline_analytics_processor.stop()

flagsmith/mappers.py

@@ -12,7 +12,6 @@
     StrValueSegmentCondition,
 )
 from flag_engine.result.types import SegmentResult
-from flag_engine.segments.types import ContextValue
 
 from flagsmith.api.types import (
     EnvironmentModel,
@@ -26,7 +25,7 @@
     SDKEvaluationContext,
     SegmentMetadata,
     StreamEvent,
-    TraitConfig,
+    TraitMapping,
 )
 from flagsmith.utils.datetime import fromisoformat
 
@@ -75,31 +74,27 @@ def map_environment_document_to_environment_updated_at(
     return updated_at.astimezone(tz=timezone.utc)
 
 
+def resolve_trait_values(
+    traits: typing.Optional[TraitMapping],
+) -> typing.Optional[typing.Dict[str, typing.Any]]:
+    if not traits:
+        return None
+    return {
+        key: (val["value"] if isinstance(val, dict) else val)
+        for key, val in traits.items()
+    }
+
+
 def map_context_and_identity_data_to_context(
     context: SDKEvaluationContext,
     identifier: str,
-    traits: typing.Optional[
-        typing.Mapping[
-            str,
-            typing.Union[
-                ContextValue,
-                TraitConfig,
-            ],
-        ]
-    ],
+    traits: typing.Optional[TraitMapping] = None,
 ) -> SDKEvaluationContext:
     return {
         **context,
         "identity": {
             "identifier": identifier,
-            "traits": {
-                trait_key: (
-                    trait_value_or_config["value"]
-                    if isinstance(trait_value_or_config, dict)
-                    else trait_value_or_config
-                )
-                for trait_key, trait_value_or_config in (traits or {}).items()
-            },
+            "traits": resolve_trait_values(traits) or {},
         },
     }