Don't look up K_CE_OVERRIDES in every call.

Author: pantierra

eoapi_notifier/outputs/cloudevents.py

@@ -92,6 +92,9 @@ def __init__(self, config: CloudEventsConfig) -> None:
         super().__init__(config)
         self.config: CloudEventsConfig = config
         self._client: httpx.AsyncClient | None = None
+        # Cache for K_CE_OVERRIDES to avoid repeated JSON parsing
+        self._cached_k_ce_overrides: str | None = None
+        self._cached_ce_extensions: dict[str, str] = {}
 
     async def start(self) -> None:
         """Start the HTTP client."""
@@ -197,22 +200,42 @@ async def send_event(self, event: NotificationEvent) -> bool:
             )
             return False
 
+    def _get_cached_ce_extensions(self) -> dict[str, str]:
+        """Get cached CloudEvents extensions from K_CE_OVERRIDES.
+
+        Parsing only happens if the environment variable has changed.
+        """
+        current_overrides = os.getenv("K_CE_OVERRIDES")
+
+        # If the environment variable hasn't changed, return cached result
+        if current_overrides == self._cached_k_ce_overrides:
+            return self._cached_ce_extensions
+
+        # Environment variable changed or first time - re-parse
+        self._cached_k_ce_overrides = current_overrides
+
+        if current_overrides:
+            try:
+                overrides_data = json.loads(current_overrides)
+                self._cached_ce_extensions = overrides_data.get("extensions", {})
+            except json.JSONDecodeError:
+                self.logger.warning(
+                    "Invalid K_CE_OVERRIDES JSON, ignoring: %s", current_overrides
+                )
+                self._cached_ce_extensions = {}
+        else:
+            self._cached_ce_extensions = {}
+
+        return self._cached_ce_extensions
+
     def _convert_to_cloudevent(self, event: NotificationEvent) -> CloudEvent:
         """Convert NotificationEvent to CloudEvent."""
         # Use config values which now include environment overrides
         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
-                )
+        # Get cached KNative CE overrides, parsing only if environment variable changed
+        ce_extensions = self._get_cached_ce_extensions()
 
         # Map operation to event type suffix
         operation_map = {"INSERT": "created", "UPDATE": "updated", "DELETE": "deleted"}