feat: Support SDK metrics

flagsmith/__init__.py

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

flagsmith/flagsmith.py

@@ -11,6 +11,7 @@
 from flag_engine.identities.traits.types import TraitValue
 from flag_engine.segments.evaluator import get_identity_segments
 from requests.adapters import HTTPAdapter
+from requests.utils import default_user_agent
 from urllib3 import Retry
 
 from flagsmith.analytics import AnalyticsProcessor
@@ -19,13 +20,20 @@
 from flagsmith.offline_handlers import BaseOfflineHandler
 from flagsmith.polling_manager import EnvironmentDataPollingManager
 from flagsmith.streaming_manager import EventStreamManager, StreamEvent
-from flagsmith.types import JsonType, TraitConfig, TraitMapping
+from flagsmith.types import (
+    ApplicationMetadata,
+    JsonType,
+    TraitConfig,
+    TraitMapping,
+)
 from flagsmith.utils.identities import generate_identity_data
+from flagsmith.version import __version__
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_API_URL = "https://edge.api.flagsmith.com/api/v1/"
 DEFAULT_REALTIME_API_URL = "https://realtime.flagsmith.com/"
+DEFAULT_USER_AGENT = f"flagsmith-python-client/{__version__} " + default_user_agent()
 
 
 class Flagsmith:
@@ -61,6 +69,7 @@ def __init__(
         offline_mode: bool = False,
         offline_handler: typing.Optional[BaseOfflineHandler] = None,
         enable_realtime_updates: bool = False,
+        application_metadata: typing.Optional[ApplicationMetadata] = None,
     ):
         """
         :param environment_key: The environment key obtained from Flagsmith interface.
@@ -88,6 +97,7 @@ def __init__(
             document from another source when in offline_mode. Works in place of
             default_flag_handler if offline_mode is not set and using remote evaluation.
         :param enable_realtime_updates: Use real-time functionality via SSE as opposed to polling the API
+        :param application_metadata: Optional metadata about the client application.
         """
 
         self.offline_mode = offline_mode
@@ -122,7 +132,11 @@ def __init__(
 
             self.session = requests.Session()
             self.session.headers.update(
-                **{"X-Environment-Key": environment_key}, **(custom_headers or {})
+                self._get_headers(
+                    environment_key=environment_key,
+                    application_metadata=application_metadata,
+                    custom_headers=custom_headers,
+                )
             )
             self.session.proxies.update(proxies or {})
             retries = retries or Retry(total=3, backoff_factor=0.1)
@@ -275,6 +289,24 @@ def update_environment(self) -> None:
                     identity.identifier: identity for identity in overrides
                 }
 
+    def _get_headers(
+        self,
+        environment_key: str,
+        application_metadata: typing.Optional[ApplicationMetadata],
+        custom_headers: typing.Optional[typing.Dict[str, typing.Any]],
+    ) -> typing.Dict[str, str]:
+        headers = {
+            "X-Environment-Key": environment_key,
+            "User-Agent": DEFAULT_USER_AGENT,
+        }
+        if application_metadata:
+            if name := application_metadata.get("name"):
+                headers["Flagsmith-Application-Name"] = name
+            if version := application_metadata.get("version"):
+                headers["Flagsmith-Application-Version"] = version
+        headers.update(custom_headers or {})
+        return headers
+
     def _get_environment_from_api(self) -> EnvironmentModel:
         environment_data = self._get_json_response(self.environment_url, method="GET")
         return EnvironmentModel.model_validate(environment_data)

flagsmith/types.py

@@ -1,7 +1,7 @@
 import typing
 
 from flag_engine.identities.traits.types import TraitValue
-from typing_extensions import TypeAlias
+from typing_extensions import NotRequired, TypeAlias
 
 _JsonScalarType: TypeAlias = typing.Union[
     int,
@@ -23,3 +23,8 @@ class TraitConfig(typing.TypedDict):
 
 
 TraitMapping: TypeAlias = typing.Mapping[str, typing.Union[TraitValue, TraitConfig]]
+
+
+class ApplicationMetadata(typing.TypedDict):
+    name: NotRequired[str]
+    version: NotRequired[str]

flagsmith/version.py

@@ -0,0 +1,3 @@
+from importlib.metadata import version
+
+__version__ = version("flagsmith")

tests/test_flagsmith.py

@@ -11,7 +11,7 @@
 from pytest_mock import MockerFixture
 from responses import matchers
 
-from flagsmith import Flagsmith
+from flagsmith import Flagsmith, __version__
 from flagsmith.exceptions import (
     FlagsmithAPIError,
     FlagsmithFeatureDoesNotExistError,
@@ -717,3 +717,96 @@ def test_custom_feature_error_raised_when_invalid_feature(
     with pytest.raises(FlagsmithFeatureDoesNotExistError):
         # When
         flags.is_feature_enabled("non-existing-feature")
+
+
+@pytest.fixture
+def default_headers() -> typing.Dict[str, str]:
+    return {
+        "User-Agent": f"flagsmith-python-client/{__version__} python-requests/2.32.4",
+        "Accept-Encoding": "gzip, deflate",
+        "Accept": "*/*",
+        "Connection": "keep-alive",
+    }
+
+
+@pytest.mark.parametrize(
+    "kwargs,expected_headers",
+    [
+        (
+            {
+                "environment_key": "test-key",
+                "application_metadata": {"name": "test-app", "version": "1.0.0"},
+            },
+            {
+                "Flagsmith-Application-Name": "test-app",
+                "Flagsmith-Application-Version": "1.0.0",
+                "X-Environment-Key": "test-key",
+            },
+        ),
+        (
+            {
+                "environment_key": "test-key",
+                "application_metadata": {"name": "test-app"},
+            },
+            {
+                "Flagsmith-Application-Name": "test-app",
+                "X-Environment-Key": "test-key",
+            },
+        ),
+        (
+            {
+                "environment_key": "test-key",
+                "application_metadata": {"version": "1.0.0"},
+            },
+            {
+                "Flagsmith-Application-Version": "1.0.0",
+                "X-Environment-Key": "test-key",
+            },
+        ),
+        (
+            {
+                "environment_key": "test-key",
+                "application_metadata": {"version": "1.0.0"},
+                "custom_headers": {"X-Custom-Header": "CustomValue"},
+            },
+            {
+                "Flagsmith-Application-Version": "1.0.0",
+                "X-Environment-Key": "test-key",
+                "X-Custom-Header": "CustomValue",
+            },
+        ),
+        (
+            {
+                "environment_key": "test-key",
+                "application_metadata": None,
+                "custom_headers": {"X-Custom-Header": "CustomValue"},
+            },
+            {
+                "X-Environment-Key": "test-key",
+                "X-Custom-Header": "CustomValue",
+            },
+        ),
+        (
+            {"environment_key": "test-key"},
+            {
+                "X-Environment-Key": "test-key",
+            },
+        ),
+    ],
+)
+@responses.activate()
+def test_flagsmith__init__expected_headers_sent(
+    default_headers: typing.Dict[str, str],
+    kwargs: typing.Dict[str, typing.Any],
+    expected_headers: typing.Dict[str, str],
+) -> None:
+    # Given
+    flagsmith = Flagsmith(**kwargs)
+    responses.add(method="GET", url=flagsmith.environment_flags_url, body="{}")
+
+    # When
+    flagsmith.get_environment_flags()
+
+    # Then
+    headers = responses.calls[0].request.headers
+    assert headers == {**default_headers, **expected_headers}