feat: impact metrics

Author: kwasniew

UnleashClient/__init__.py

@@ -35,6 +35,7 @@
     SDK_NAME,
     SDK_VERSION,
 )
+from UnleashClient.environment_resolver import extract_environment_from_headers
 from UnleashClient.events import (
     BaseEvent,
     UnleashEvent,
@@ -206,8 +207,14 @@ def __init__(
         self.fl_job: Job = None
         self.metric_job: Job = None
         self.engine = UnleashEngine()
+
+        impact_metrics_environment = self.unleash_environment
+        extracted_env = extract_environment_from_headers(self.unleash_custom_headers)
+        if extracted_env:
+            impact_metrics_environment = extracted_env
+
         self._impact_metrics = ImpactMetrics(
-            self.engine, self.unleash_app_name, self.unleash_environment
+            self.engine, self.unleash_app_name, impact_metrics_environment
         )
 
         self.cache = cache or FileCache(

UnleashClient/environment_resolver.py

@@ -0,0 +1,26 @@
+from typing import Dict, Optional
+
+
+def extract_environment_from_headers(
+    headers: Optional[Dict[str, str]],
+) -> Optional[str]:
+    if not headers:
+        return None
+
+    auth_key = next(
+        (key for key in headers if key.lower() == "authorization"),
+        None,
+    )
+    if not auth_key:
+        return None
+
+    auth_value = headers.get(auth_key)
+    if not auth_value:
+        return None
+
+    _, sep, after_colon = auth_value.partition(":")
+    if not sep:
+        return None
+
+    environment, _, _ = after_colon.partition(".")
+    return environment or None

tests/unit_tests/test_environment_resolver.py

@@ -0,0 +1,35 @@
+from UnleashClient.environment_resolver import extract_environment_from_headers
+
+
+def test_valid_headers():
+    custom_headers = {
+        "Authorization": "project:environment.hash",
+        "Content-Type": "application/json",
+    }
+
+    result = extract_environment_from_headers(custom_headers)
+    assert result == "environment"
+
+
+def test_case_insensitive_header_keys():
+    custom_headers = {
+        "AUTHORIZATION": "project:environment.hash",
+        "Content-Type": "application/json",
+    }
+
+    result = extract_environment_from_headers(custom_headers)
+    assert result == "environment"
+
+
+def test_authorization_header_not_present():
+    result = extract_environment_from_headers({})
+    assert result is None
+
+
+def test_environment_part_is_empty():
+    custom_headers = {
+        "Authorization": "project:.hash",
+    }
+
+    result = extract_environment_from_headers(custom_headers)
+    assert result is None