refactoring, removing useless changes

Author: Avi-Robusta

src/robusta/core/reporting/base.py

@@ -149,14 +149,6 @@ def __init__(
 
     def __str__(self):
         return f"annotations: {self.annotations} Enrichment: {self.blocks} "
-    
-    def to_dict(self):
-        return {
-            "blocks": [block.dict() for block in self.blocks],
-            "annotations": self.annotations,
-            "enrichment_type": self.enrichment_type,
-            "title": self.title,
-        }
 
 
 class FilterableScopeMatcher(BaseScopeMatcher):
@@ -412,31 +404,3 @@ def __calculate_fingerprint(subject: FindingSubject, source: FindingSource, aggr
         # if not, generate with logic similar to alertmanager
         s = f"{subject.subject_type},{subject.name},{subject.namespace},{subject.node},{source.value}{aggregation_key}"
         return hashlib.sha256(s.encode()).hexdigest()
-    
-    def to_json(self):
-        return {
-            "title": self.title,
-            "aggregation_key": self.aggregation_key,
-            "severity": self.severity.name,
-            "source": self.source.name,
-            "description": self.description,
-            "subject": {
-                "name": self.subject.name,
-                "subject_type": self.subject.subject_type.value,
-                "namespace": self.subject.namespace,
-                "node": self.subject.node,
-                "container": self.subject.container,
-                "labels": self.subject.labels,
-                "annotations": self.subject.annotations,
-            },
-            "finding_type": self.finding_type.name,
-            "failure": self.failure,
-            "creation_date": self.creation_date,
-            "fingerprint": self.fingerprint,
-            "starts_at": self.starts_at,
-            "ends_at": self.ends_at,
-            "add_silence_url": self.add_silence_url,
-            "silence_labels": self.silence_labels,
-            "enrichments": [enrichment.to_dict() for enrichment in self.enrichments],
-            "links": [link.dict() for link in self.links],
-        }

src/robusta/core/sinks/slack/__init__.py

@@ -1,4 +1,5 @@
 from robusta.core.sinks.slack.slack_sink_params import SlackSinkParams, SlackSinkConfigWrapper
 from robusta.core.sinks.slack.slack_sink import SlackSink
 
+# to prevent circular imports in SlackSender, SlackSinkParams and SlackSinkPreviewParams
 __all__ = ["SlackSink", "SlackSinkParams", "SlackSinkConfigWrapper"]

src/robusta/core/sinks/slack/preview/slack_sink_preview_params.py

@@ -1,38 +1,36 @@
 from robusta.core.sinks.sink_base_params import SinkBaseParams
 from robusta.core.sinks.sink_config import SinkConfigBase
 from robusta.core.sinks.slack.slack_sink_params import SlackSinkParams
-from enum import Enum
-from typing import Optional, Dict, Any
-
-
-class SlackTemplateStyle(str, Enum):
-    DEFAULT = "default"
-    LEGACY = "legacy"
+from typing import Optional, Dict
+from pydantic import validator
 
 
 class SlackSinkPreviewParams(SlackSinkParams):
+    #TODO: improve the SlackSinkPreviewParams so the slack_custom_templates can be defined once globally and
+    # only a template name needs to be passed to each channel in the config
     slack_custom_templates: Optional[Dict[str, str]] = None  # Template name -> custom template content
-    template_name: Optional[str] = None
 
-    def get_effective_template_name(self) -> str:
+    @validator('slack_custom_templates')
+    def check_one_item(cls, v):
+        if v is not None and len(v) != 1:
+            raise ValueError("slack_custom_templates must contain exactly one key-value pair")
+        return v
+
+    def get_template_name(self) -> str:
         """
-        Returns the template name to use for this sink. If template_name is set, use it.
-        Otherwise, use 'legacy.j2' if template_style is legacy, else 'header.j2'.
+        Returns the template name to use for this sink. If slack_custom_templates is set, use the first one.
+        Otherwise, use 'header.j2'.
         """
-        if self.slack_custom_templates and len(self.slack_custom_templates) == 1:
+        if self.slack_custom_templates:
             return next(iter(self.slack_custom_templates))
         return "header.j2"
 
     def get_custom_template(self) -> Optional[str]:
-        """Get the custom template for the current template style"""
+        """Get the custom template if defined"""
         if not self.slack_custom_templates:
             return None
 
-        template_name = self.get_effective_template_name()
-        if template_name not in self.slack_custom_templates:
-            return None
-
-        return self.slack_custom_templates[template_name]
+        return next(iter(self.slack_custom_templates.values()))
 
 
 class SlackSinkPreviewConfigWrapper(SinkConfigBase):

src/robusta/core/sinks/slack/slack_sink_params.py

@@ -1,6 +1,7 @@
 from robusta.core.sinks.sink_base_params import SinkBaseParams
 from robusta.core.sinks.sink_config import SinkConfigBase
 from robusta.core.sinks.common import ChannelTransformer
+
 from typing import Optional
 from pydantic import validator
 

src/robusta/core/sinks/slack/templates/README.md

@@ -78,31 +78,16 @@ def render_to_blocks(self, template: Template, context: Dict[str, Any]) -> List[
             return []
 
     def render_custom_template_to_blocks(self, custom_template: str, context: Dict[str, Any]) -> List[Dict[str, Any]]:
-        """
-        Render a custom Jinja template string to Slack blocks.
-        Args:
-            custom_template: Jinja template string
-            context: Dictionary of variables to pass to the template
-        Returns:
-            List of Slack block objects (dictionaries)
-        """
         try:
             template = Template(custom_template)
             return self.render_to_blocks(template, context)
         except Exception as e:
             logging.error(f"Error rendering custom template: {e}")
-            return []
+            return self.render_default_template_to_blocks(context)
 
-    def render_file_template_to_blocks(self, template_name: str, context: Dict[str, Any]) -> List[Dict[str, Any]]:
-        """
-        Render a file-based Jinja template to Slack blocks.
-        Args:
-            template_name: The name of the file-based template (e.g., "header.j2")
-            context: Dictionary of variables to pass to the template
-        Returns:
-            List of Slack block objects (dictionaries)
-        """
-        template = self.get_template(template_name)
+    def render_default_template_to_blocks(self, context: Dict[str, Any]) -> List[Dict[str, Any]]:
+        DEFAULT_TEMPLATE_NAME="header.j2"
+        template = self.get_template(DEFAULT_TEMPLATE_NAME)
         return self.render_to_blocks(template, context)
 
 

src/robusta/core/sinks/slack/templates/__init__.py

@@ -509,7 +509,7 @@ def __create_finding_header_preview(
         }
 
         # Determine the template name to use
-        template_name = sink_params.get_effective_template_name() if sink_params else "header.j2"
+        template_name = sink_params.get_template_name() if sink_params else "header.j2"
 
         # Get the custom template for this template name, if any
         custom_template = sink_params.get_custom_template() if sink_params else None
@@ -518,7 +518,7 @@ def __create_finding_header_preview(
         if custom_template:
             return template_loader.render_custom_template_to_blocks(custom_template, template_context)
         else:
-            return template_loader.render_file_template_to_blocks(template_name, template_context)
+            return template_loader.render_default_template_to_blocks(template_context)
 
     def __create_finding_header(
         self, finding: Finding, status: FindingStatus, platform_enabled: bool, include_investigate_link: bool
@@ -676,12 +676,15 @@ def send_finding_to_slack(
         thread_ts: str = None,
     ) -> str:
         if self.is_preview:
-            return self.__send_finding_to_slack_preview(
-                finding=finding,
-                sink_params=sink_params,
-                platform_enabled=platform_enabled,
-                thread_ts=thread_ts
-            )
+            try:
+                return self.__send_finding_to_slack_preview(
+                    finding=finding,
+                    sink_params=sink_params,
+                    platform_enabled=platform_enabled,
+                    thread_ts=thread_ts
+                )
+            except Exception:
+                logging.exception("Failed to render slack preview template, defaulting to legacy slack output")
         return self.__send_finding_to_slack(
             finding=finding,
             sink_params=sink_params,
@@ -793,14 +796,11 @@ def __send_finding_to_slack_preview(
             FindingStatus.RESOLVED if finding.title.startswith("[RESOLVED]") else FindingStatus.FIRING
         )
 
-        # Get JIRA-style header blocks
         if finding.title:
             header_blocks = self.__create_finding_header_preview(finding, status, platform_enabled,
                                                          sink_params.investigate_link, sink_params)
 
-        # Description handling - moved above the buttons
         if finding.description:
-            # Always show description immediately after title
             description_text = finding.description
             blocks.append(MarkdownBlock(description_text))
 

src/robusta/core/sinks/slack/templates/legacy.j2

@@ -25,7 +25,7 @@
 )
 from robusta.core.model.base_params import ResourceInfo
 from robusta.core.sinks.slack.slack_sink_params import SlackSinkParams
-from robusta.core.sinks.slack.preview.slack_sink_preview_params import SlackSinkPreviewParams, SlackTemplateStyle
+from robusta.core.sinks.slack.preview.slack_sink_preview_params import SlackSinkPreviewParams
 from robusta.core.playbooks.internal.ai_integration import ask_holmes
 
 # Configure logging - set to DEBUG to see full block details