11from robusta .core .sinks .sink_base_params import SinkBaseParams
22from robusta .core .sinks .sink_config import SinkConfigBase
33from robusta .core .sinks .common import ChannelTransformer
4- from enum import Enum
5- from typing import Optional , Dict , Literal
4+ from typing import Optional
65from pydantic import validator
76
87
9- class SlackTemplateStyle (str , Enum ):
10- DEFAULT = "default"
11- LEGACY = "legacy"
12-
13-
148class SlackSinkParams (SinkBaseParams ):
159 slack_channel : str
1610 api_key : str
1711 channel_override : Optional [str ] = None
1812 max_log_file_limit_kb : int = 1000
1913 investigate_link : bool = True
2014
21- template_style : SlackTemplateStyle = SlackTemplateStyle .DEFAULT # Use "legacy" for old-style formatting
22- slack_custom_templates : Optional [Dict [str , str ]] = None # Template name -> custom template content
23- template_name : Optional [str ] = None
24-
25- def get_effective_template_name (self ) -> str :
26- """
27- Returns the template name to use for this sink. If template_name is set, use it.
28- Otherwise, use 'legacy.j2' if template_style is legacy, else 'header.j2'.
29- """
30- if self .template_name :
31- return self .template_name
32- if self .template_style == SlackTemplateStyle .LEGACY :
33- return "legacy.j2"
34- return "header.j2"
35-
36- def get_custom_template (self ) -> Optional [str ]:
37- """
38- Returns the custom template string for the effective template name, if it exists.
39- """
40- template_name = self .get_effective_template_name ()
41- if self .slack_custom_templates and template_name in self .slack_custom_templates :
42- return self .slack_custom_templates [template_name ]
43- return None
44-
4515 @classmethod
4616 def _supports_grouping (cls ):
4717 return True
@@ -59,4 +29,4 @@ class SlackSinkConfigWrapper(SinkConfigBase):
5929 slack_sink : SlackSinkParams
6030
6131 def get_params (self ) -> SinkBaseParams :
62- return self .slack_sink
32+ return self .slack_sink
0 commit comments