feat: Support SDK metrics (#136)

Author: khvn26

.github/workflows/pytest.yml

@@ -32,5 +32,9 @@ jobs:
           pip install poetry
           poetry install --with dev
 
+      - name: Check for new typing errors
+        if: ${{ matrix.python-version != '3.8' }}
+        run: poetry run mypy --strict .
+
       - name: Run Tests
         run: poetry run pytest

.pre-commit-config.yaml

@@ -1,11 +1,4 @@
 repos:
-  - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.16.1
-    hooks:
-      - id: mypy
-        args: [--strict]
-        additional_dependencies:
-          [pydantic, pytest, pytest_mock, types-requests, flagsmith-flag-engine, responses, sseclient-py]
   - repo: https://github.com/PyCQA/isort
     rev: 6.0.1
     hooks:

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

@@ -1,4 +1,5 @@
 import logging
+import sys
 import typing
 from datetime import timezone
 
@@ -11,6 +12,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 +21,24 @@
 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()
+    + f" python/{sys.version_info.major}.{sys.version_info.minor}"
+)
 
 
 class Flagsmith:
@@ -61,6 +74,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 +102,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 +137,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 +294,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")