Don't look up K_CE_OVERRIDES in every call.

Author: pantierra

eoapi_notifier/outputs/cloudevents.py

@@ -94,6 +94,26 @@ def __init__(self, config: CloudEventsConfig) -> None:
         super().__init__(config)
         self.config: CloudEventsConfig = config
         self._client: httpx.AsyncClient | None = None
+        # Parse K_CE_OVERRIDES once during initialization
+        self._ce_extensions = self._parse_k_ce_overrides()
+
+    def _parse_k_ce_overrides(self) -> dict[str, str]:
+        """Parse K_CE_OVERRIDES environment variable once during initialization."""
+        k_ce_overrides = os.getenv("K_CE_OVERRIDES")
+        if not k_ce_overrides:
+            return {}
+
+        try:
+            overrides_data = json.loads(k_ce_overrides)
+            extensions = overrides_data.get("extensions", {})
+            if isinstance(extensions, dict):
+                return {str(k): str(v) for k, v in extensions.items()}
+            return {}
+        except json.JSONDecodeError:
+            self.logger.warning(
+                "Invalid K_CE_OVERRIDES JSON, ignoring: %s", k_ce_overrides
+            )
+            return {}
 
     async def start(self) -> None:
         """Start the HTTP client."""
@@ -215,16 +235,8 @@ def _convert_to_cloudevent(self, event: NotificationEvent) -> CloudEvent:
         source = self.config.source
         event_type_base = self.config.event_type
 
-        # Apply KNative CE overrides if present
-        ce_extensions = {}
-        if k_ce_overrides := os.getenv("K_CE_OVERRIDES"):
-            try:
-                overrides_data = json.loads(k_ce_overrides)
-                ce_extensions = overrides_data.get("extensions", {})
-            except json.JSONDecodeError:
-                self.logger.warning(
-                    "Invalid K_CE_OVERRIDES JSON, ignoring: %s", k_ce_overrides
-                )
+        # Use pre-parsed KNative CE overrides
+        ce_extensions = self._ce_extensions
 
         # Map operation to event type suffix
         operation_map = {"INSERT": "created", "UPDATE": "updated", "DELETE": "deleted"}

tests/test_cloudevents_output.py

@@ -319,9 +319,11 @@ async def test_health_check(self, adapter: CloudEventsAdapter) -> None:
         },
     )
     def test_convert_to_cloudevent_with_overrides(
-        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+        self, config: CloudEventsConfig, sample_event: NotificationEvent
     ) -> None:
         """Test CloudEvent conversion with K_CE_OVERRIDES."""
+        # Create adapter after environment variable is set
+        adapter = CloudEventsAdapter(config)
         cloud_event = adapter._convert_to_cloudevent(sample_event)
 
         assert isinstance(cloud_event, CloudEvent)
@@ -330,18 +332,22 @@ def test_convert_to_cloudevent_with_overrides(
 
     @patch.dict(os.environ, {"K_CE_OVERRIDES": '{"extensions": {"number": 123}}'})
     def test_convert_to_cloudevent_with_number_override(
-        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+        self, config: CloudEventsConfig, sample_event: NotificationEvent
     ) -> None:
         """Test CloudEvent conversion with number in K_CE_OVERRIDES."""
+        # Create adapter after environment variable is set
+        adapter = CloudEventsAdapter(config)
         cloud_event = adapter._convert_to_cloudevent(sample_event)
 
         assert cloud_event["number"] == "123"  # Should be converted to string
 
     @patch.dict(os.environ, {"K_CE_OVERRIDES": "invalid-json"})
     def test_convert_to_cloudevent_invalid_overrides(
-        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+        self, config: CloudEventsConfig, sample_event: NotificationEvent
     ) -> None:
         """Test CloudEvent conversion with invalid K_CE_OVERRIDES JSON."""
+        # Create adapter after environment variable is set
+        adapter = CloudEventsAdapter(config)
         cloud_event = adapter._convert_to_cloudevent(sample_event)
 
         # Should work normally without overrides
@@ -350,9 +356,11 @@ def test_convert_to_cloudevent_invalid_overrides(
 
     @patch.dict(os.environ, {"K_CE_OVERRIDES": '{"other": "field"}'})
     def test_convert_to_cloudevent_no_extensions(
-        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+        self, config: CloudEventsConfig, sample_event: NotificationEvent
     ) -> None:
         """Test CloudEvent conversion with K_CE_OVERRIDES but no extensions field."""
+        # Create adapter after environment variable is set
+        adapter = CloudEventsAdapter(config)
         cloud_event = adapter._convert_to_cloudevent(sample_event)
 
         # Should work normally without extensions