slack

slack

Slack notification implementation using httpx.

Provides Slack webhook notifications for Marianne job events. Messages are formatted with rich Slack Block Kit formatting.

Phase 5 of Marianne implementation: Missing README features.

Classes

SlackNotifier

SlackNotifier(webhook_url=None, webhook_url_env='SLACK_WEBHOOK_URL', channel=None, username='Marianne AI Compose', icon_emoji=':musical_score:', events=None, timeout=10.0)

Slack notification implementation using webhooks.

Sends notifications to Slack channels via incoming webhooks. Supports rich formatting with Slack Block Kit attachments.

Example usage

notifier = SlackNotifier( webhook_url="https://hooks.slack.com/services/...", channel="#alerts", events={NotificationEvent.JOB_COMPLETE, NotificationEvent.JOB_FAILED}, ) await notifier.send(context)

Configuration from YAML

notifications: - type: slack on_events: [job_complete, job_failed, sheet_failed] config: webhook_url_env: SLACK_WEBHOOK_URL channel: "#marianne-alerts" username: "Marianne Bot" timeout: 10

Initialize the Slack notifier.

Parameters:

Name	Type	Description	Default
webhook_url	str | None	Direct webhook URL. If not provided, reads from env var.	None
webhook_url_env	str	Environment variable containing webhook URL.	'SLACK_WEBHOOK_URL'
channel	str | None	Override channel (optional, uses webhook default if not set).	None
username	str	Bot username displayed in Slack.	'Marianne AI Compose'
icon_emoji	str	Emoji for bot avatar.	':musical_score:'
events	set[NotificationEvent] | None	Set of events to subscribe to. Defaults to job-level events.	None
timeout	float	HTTP request timeout in seconds.	10.0

Source code in src/marianne/notifications/slack.py

def __init__(
    self,
    webhook_url: str | None = None,
    webhook_url_env: str = "SLACK_WEBHOOK_URL",
    channel: str | None = None,
    username: str = "Marianne AI Compose",
    icon_emoji: str = ":musical_score:",
    events: set[NotificationEvent] | None = None,
    timeout: float = 10.0,
) -> None:
    """Initialize the Slack notifier.

    Args:
        webhook_url: Direct webhook URL. If not provided, reads from env var.
        webhook_url_env: Environment variable containing webhook URL.
        channel: Override channel (optional, uses webhook default if not set).
        username: Bot username displayed in Slack.
        icon_emoji: Emoji for bot avatar.
        events: Set of events to subscribe to. Defaults to job-level events.
        timeout: HTTP request timeout in seconds.
    """
    # Get webhook URL from param or environment
    self._webhook_url = webhook_url or os.environ.get(webhook_url_env, "")
    self._channel = channel
    self._username = username
    self._icon_emoji = icon_emoji
    self._timeout = timeout

    self._events = events or {
        NotificationEvent.JOB_COMPLETE,
        NotificationEvent.JOB_FAILED,
        NotificationEvent.SHEET_FAILED,
    }

    self._client: httpx.AsyncClient | None = None
    self._warned_no_webhook = False

Attributes

subscribed_events property

subscribed_events

Events this notifier is registered to receive.

Functions

from_config classmethod

from_config(on_events, config=None)

Create SlackNotifier from YAML configuration.

Parameters:

Name	Type	Description	Default
on_events	list[str]	List of event name strings from config.	required
config	dict[str, Any] | None	Optional dict with Slack-specific settings: - webhook_url: Direct webhook URL - webhook_url_env: Env var for webhook URL (default: SLACK_WEBHOOK_URL) - channel: Override channel - username: Bot username - icon_emoji: Bot avatar emoji - timeout: Request timeout in seconds	None

Returns:

Type	Description
SlackNotifier	Configured SlackNotifier instance.

Example

notifier = SlackNotifier.from_config( on_events=["job_complete", "job_failed"], config={ "webhook_url_env": "MY_SLACK_WEBHOOK", "channel": "#alerts", }, )

Source code in src/marianne/notifications/slack.py

@classmethod
def from_config(
    cls,
    on_events: list[str],
    config: dict[str, Any] | None = None,
) -> "SlackNotifier":
    """Create SlackNotifier from YAML configuration.

    Args:
        on_events: List of event name strings from config.
        config: Optional dict with Slack-specific settings:
            - webhook_url: Direct webhook URL
            - webhook_url_env: Env var for webhook URL (default: SLACK_WEBHOOK_URL)
            - channel: Override channel
            - username: Bot username
            - icon_emoji: Bot avatar emoji
            - timeout: Request timeout in seconds

    Returns:
        Configured SlackNotifier instance.

    Example:
        notifier = SlackNotifier.from_config(
            on_events=["job_complete", "job_failed"],
            config={
                "webhook_url_env": "MY_SLACK_WEBHOOK",
                "channel": "#alerts",
            },
        )
    """
    config = config or {}

    # Convert string event names to NotificationEvent enums
    events: set[NotificationEvent] = set()
    for event_name in on_events:
        try:
            normalized = event_name.upper()
            events.add(NotificationEvent[normalized])
        except KeyError:
            _logger.warning("unknown_notification_event", event_name=event_name)

    return cls(
        webhook_url=config.get("webhook_url"),
        webhook_url_env=config.get("webhook_url_env", "SLACK_WEBHOOK_URL"),
        channel=config.get("channel"),
        username=config.get("username", "Marianne AI Compose"),
        icon_emoji=config.get("icon_emoji", ":musical_score:"),
        events=events if events else None,
        timeout=config.get("timeout", 10.0),
    )

send async

send(context)

Send a Slack notification.

Posts to the configured Slack webhook with rich formatting. Fails gracefully if webhook is unavailable or request fails.

Parameters:

Name	Type	Description	Default
context	NotificationContext	Notification context with event details.	required

Returns:

Type	Description
bool	True if notification was sent, False if unavailable or failed.

Source code in src/marianne/notifications/slack.py

async def send(self, context: NotificationContext) -> bool:
    """Send a Slack notification.

    Posts to the configured Slack webhook with rich formatting.
    Fails gracefully if webhook is unavailable or request fails.

    Args:
        context: Notification context with event details.

    Returns:
        True if notification was sent, False if unavailable or failed.
    """
    if not self._webhook_url:
        if not self._warned_no_webhook:
            _logger.warning(
                "Slack webhook URL not configured. "
                "Set webhook_url or SLACK_WEBHOOK_URL environment variable."
            )
            self._warned_no_webhook = True
        return False

    if context.event not in self._events:
        # Not subscribed to this event
        return True

    try:
        client = await self._get_client()
        payload = self._build_payload(context)

        response = await client.post(
            self._webhook_url,
            json=payload,
        )

        if response.status_code == 200:
            _logger.debug("slack_notification_sent", title=context.format_title())
            return True

        _logger.warning(
            "slack_webhook_error",
            status_code=response.status_code,
            body=response.text[:200],
        )
        return False

    except httpx.TimeoutException:
        _logger.warning("Slack notification timed out")
        return False
    except httpx.RequestError as e:
        _logger.warning("slack_notification_failed", error=str(e))
        return False
    except Exception as e:
        _logger.warning("slack_notification_unexpected_error", error=str(e), exc_info=True)
        return False

close async

close()

Clean up HTTP client resources.

Called when the NotificationManager is shutting down.

Source code in src/marianne/notifications/slack.py

async def close(self) -> None:
    """Clean up HTTP client resources.

    Called when the NotificationManager is shutting down.
    """
    if self._client is not None and not self._client.is_closed:
        await self._client.aclose()
        self._client = None

MockSlackNotifier

MockSlackNotifier(events=None)

Mock Slack notifier for testing.

Records all notifications sent without making HTTP calls. Useful for testing notification integration.

Initialize mock notifier.

Parameters:

Name	Type	Description	Default
events	set[NotificationEvent] | None	Set of events to subscribe to.	None

Source code in src/marianne/notifications/slack.py

def __init__(
    self,
    events: set[NotificationEvent] | None = None,
) -> None:
    """Initialize mock notifier.

    Args:
        events: Set of events to subscribe to.
    """
    self._events = events or {
        NotificationEvent.JOB_COMPLETE,
        NotificationEvent.JOB_FAILED,
        NotificationEvent.SHEET_FAILED,
    }
    self.sent_notifications: list[NotificationContext] = []
    self.sent_payloads: list[dict[str, Any]] = []
    self._fail_next = False

Attributes

subscribed_events property

subscribed_events

Events this notifier handles.

Functions

set_fail_next

set_fail_next(should_fail=True)

Configure the next send() call to fail.

Parameters:

Name	Type	Description	Default
should_fail	bool	If True, next send() returns False.	True

Source code in src/marianne/notifications/slack.py

def set_fail_next(self, should_fail: bool = True) -> None:
    """Configure the next send() call to fail.

    Args:
        should_fail: If True, next send() returns False.
    """
    self._fail_next = should_fail

send async

send(context)

Record notification without making HTTP call.

Parameters:

Name	Type	Description	Default
context	NotificationContext	Notification context.	required

Returns:

Type	Description
bool	True unless set_fail_next was called.

Source code in src/marianne/notifications/slack.py

async def send(self, context: NotificationContext) -> bool:
    """Record notification without making HTTP call.

    Args:
        context: Notification context.

    Returns:
        True unless set_fail_next was called.
    """
    if self._fail_next:
        self._fail_next = False
        return False

    self.sent_notifications.append(context)
    # Build payload like real notifier would
    notifier = SlackNotifier(webhook_url="https://mock")
    self.sent_payloads.append(notifier._build_payload(context))
    return True

close async

close()

Clear recorded notifications.

Source code in src/marianne/notifications/slack.py

async def close(self) -> None:
    """Clear recorded notifications."""
    self.sent_notifications.clear()
    self.sent_payloads.clear()

get_notification_count

get_notification_count()

Get number of recorded notifications.

Source code in src/marianne/notifications/slack.py

def get_notification_count(self) -> int:
    """Get number of recorded notifications."""
    return len(self.sent_notifications)

get_notifications_for_event

get_notifications_for_event(event)

Get all notifications for a specific event type.

Parameters:

Name	Type	Description	Default
event	NotificationEvent	Event type to filter by.	required

Returns:

Type	Description
list[NotificationContext]	List of matching notification contexts.

Source code in src/marianne/notifications/slack.py

def get_notifications_for_event(
    self, event: NotificationEvent
) -> list[NotificationContext]:
    """Get all notifications for a specific event type.

    Args:
        event: Event type to filter by.

    Returns:
        List of matching notification contexts.
    """
    return [n for n in self.sent_notifications if n.event == event]

Functions