separate slack code for easy reusage outside a script class

shub_workflow/contrib/slack.py

@@ -1,6 +1,6 @@
 import logging
 from pprint import pformat
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from spidermon.contrib.actions.slack import SlackMessageManager
 
@@ -10,38 +10,47 @@
 LOG = logging.getLogger(__name__)
 
 
+class SlackSender:
+    def __init__(self, project_settings):
+        super().__init__()
+        self.slack_handler = SlackMessageManager(
+            fake=project_settings.getbool("SPIDERMON_SLACK_FAKE"),
+            sender_token=project_settings.get("SPIDERMON_SLACK_SENDER_TOKEN"),
+            sender_name=project_settings.get("SPIDERMON_SLACK_SENDER_NAME"),
+        )
+        self.recipients = project_settings.get("SPIDERMON_SLACK_RECIPIENTS")
+
+    def send_slack_messages(self, messages: List[str], subject: str):
+        message: Dict[str, Any] = dict()
+        title = f"{self.slack_handler.sender_name} | {subject}"
+        message["title"] = title
+        message["failure_reasons"] = "\n".join(messages)
+        job_key = resolve_shub_jobkey()
+        if job_key:
+            message["job_link"] = f"https://app.zyte.com/p/{job_key}"
+        text = (
+            f"{title}\n\n"
+            "Alert Reasons:\n"
+            f"{message['failure_reasons']}\n\n"
+            f"Job Link: {message.get('job_link', 'N/A')}"
+        )
+        if self.slack_handler.fake:
+            message["failure_reasons"] = messages
+            LOG.info(pformat(message))
+        else:
+            self.slack_handler.send_message(self.recipients, text)
+
+
 class SlackMixin(AlertSenderMixin):
     """
     A class for adding slack alert capabilities to a shub_workflow class.
     """
 
     def __init__(self):
         super().__init__()
-        self.slack_handler = SlackMessageManager(
-            fake=self.project_settings.getbool("SPIDERMON_SLACK_FAKE"),
-            sender_token=self.project_settings.get("SPIDERMON_SLACK_SENDER_TOKEN"),
-            sender_name=self.project_settings.get("SPIDERMON_SLACK_SENDER_NAME"),
-        )
-        self.register_sender_method(self.send_slack_message)
+        self.register_sender_method(self.send_slack_queued_messages)
+        self.sender = SlackSender(self.project_settings)
 
-    def send_slack_message(self):
+    def send_slack_queued_messages(self):
         if self.messages:
-            message: Dict[str, Any] = dict()
-            title = f"{self.slack_handler.sender_name} | {self.args.subject}"
-            message["title"] = title
-            message["failure_reasons"] = "\n".join(self.messages)
-            job_key = resolve_shub_jobkey()
-            if job_key:
-                message["job_link"] = f"https://app.zyte.com/p/{job_key}"
-            text = (
-                f"{title}\n\n"
-                "Alert Reasons:\n"
-                f"{message['failure_reasons']}\n\n"
-                f"Job Link: {message.get('job_link', 'N/A')}"
-            )
-            to = self.project_settings.get("SPIDERMON_SLACK_RECIPIENTS")
-            if self.slack_handler.fake:
-                message["failure_reasons"] = self.messages
-                LOG.info(pformat(message))
-            else:
-                self.slack_handler.send_message(to, text)
+            self.sender.send_slack_messages(self.messages, self.args.subject)