"""Slack connector.""" from typing import Optional from slack_sdk import WebClient from notifications.config import BULK_APP_SLACK_CHANNEL_ID, BULK_APP_SLACK_TOKEN def get_slack_client(token: str) -> WebClient: """Return a Slack WebClient instance.""" return WebClient(token=token) def send_message( *, channel_id: str, token: str, body: str, color: str, thread_ts: Optional[str] = None ) -> str: """Send a message to a Slack channel. Returns: The timestamp of the sent message, which is used as a unique identifier. Raises: ValueError: If no timestamp is returned from the Slack API. """ client = get_slack_client(token) response = client.chat_postMessage( channel=channel_id, unfurl_links=False, unfurl_media=False, thread_ts=thread_ts, attachments=[{'color': color, 'text': body}], ) ts = response.get('ts') if not ts: raise ValueError('No timestamp returned from Slack API') return ts def send_bulk_message(*, body: str, color: str, thread_ts: Optional[str] = None) -> str: """Send a message to the bulk app Slack channel.""" return send_message( channel_id=BULK_APP_SLACK_CHANNEL_ID, body=body, token=BULK_APP_SLACK_TOKEN, color=color, thread_ts=thread_ts, )