feat/add-trackevent-and-pipeline-config-in-clients

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/flagsmith.py

@@ -8,7 +8,11 @@
 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,
@@ -63,6 +67,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 +113,7 @@ 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
 
@@ -175,6 +181,13 @@ def __init__(
                     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
         # method calls, update the environment manually.
@@ -290,6 +303,36 @@ 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=self._resolve_traits(traits),
+            metadata=metadata,
+        )
+
+    @staticmethod
+    def _resolve_traits(
+        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 update_environment(self) -> None:
         try:
             environment_data = self._get_json_response(
@@ -345,6 +388,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 +412,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=self._resolve_traits(traits),
         )
 
     def _get_environment_flags_from_api(self) -> Flags:
@@ -379,6 +426,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 +459,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=self._resolve_traits(traits),
             )
         except FlagsmithAPIError:
             if self.offline_handler:
@@ -443,3 +494,6 @@ def __del__(self) -> None:
 
         if hasattr(self, "event_stream_thread"):
             self.event_stream_thread.stop()
+
+        if hasattr(self, "_pipeline_analytics_processor") and self._pipeline_analytics_processor:
+            self._pipeline_analytics_processor.stop()

tests/test_flagsmith.py

@@ -9,6 +9,7 @@
 from responses import matchers
 
 from flagsmith import Flagsmith, __version__
+from flagsmith.analytics import PipelineAnalyticsConfig
 from flagsmith.api.types import EnvironmentModel
 from flagsmith.exceptions import (
     FlagsmithAPIError,
@@ -915,3 +916,84 @@ def test_flagsmith__init__expected_headers_sent(
         "Connection": "keep-alive",
         **expected_headers,
     }
+
+
+def test_track_event_raises_without_config(api_key: str) -> None:
+    flagsmith = Flagsmith(environment_key=api_key)
+    with pytest.raises(ValueError, match="Pipeline analytics is not configured"):
+        flagsmith.track_event("purchase")
+
+
+def test_track_event_calls_pipeline_processor(
+    mocker: MockerFixture, api_key: str
+) -> None:
+    config = PipelineAnalyticsConfig(analytics_server_url="http://test/")
+    flagsmith = Flagsmith(environment_key=api_key, pipeline_analytics_config=config)
+
+    mock_record = mocker.patch.object(
+        flagsmith._pipeline_analytics_processor, "record_custom_event"
+    )
+
+    flagsmith.track_event(
+        "purchase",
+        identity_identifier="user1",
+        traits={"plan": "premium"},
+        metadata={"amount": 99},
+    )
+
+    mock_record.assert_called_once_with(
+        event_name="purchase",
+        identity_identifier="user1",
+        traits={"plan": "premium"},
+        metadata={"amount": 99},
+    )
+
+
+@responses.activate()
+def test_get_flag_records_evaluation_event_via_flagsmith(
+    mocker: MockerFixture, api_key: str, flags_json: str
+) -> None:
+    config = PipelineAnalyticsConfig(analytics_server_url="http://test/")
+    flagsmith = Flagsmith(environment_key=api_key, pipeline_analytics_config=config)
+
+    mock_record = mocker.patch.object(
+        flagsmith._pipeline_analytics_processor, "record_evaluation_event"
+    )
+
+    responses.add(method="GET", url=flagsmith.environment_flags_url, body=flags_json)
+    flags = flagsmith.get_environment_flags()
+    flags.get_flag("some_feature")
+
+    mock_record.assert_called_once_with(
+        flag_key="some_feature",
+        enabled=True,
+        value="some-value",
+        identity_identifier=None,
+        traits=None,
+    )
+
+
+@responses.activate()
+def test_get_identity_flags_passes_identity_and_traits(
+    mocker: MockerFixture, api_key: str, identities_json: str
+) -> None:
+    config = PipelineAnalyticsConfig(analytics_server_url="http://test/")
+    flagsmith = Flagsmith(environment_key=api_key, pipeline_analytics_config=config)
+
+    mock_record = mocker.patch.object(
+        flagsmith._pipeline_analytics_processor, "record_evaluation_event"
+    )
+
+    responses.add(method="POST", url=flagsmith.identities_url, body=identities_json)
+    flags = flagsmith.get_identity_flags(
+        "user123", traits={"plan": "premium"}
+    )
+    flags.get_flag("some_feature")
+
+    mock_record.assert_called_once_with(
+        flag_key="some_feature",
+        enabled=True,
+        value="some-value",
+        identity_identifier="user123",
+        traits={"plan": "premium"},
+    )