orchestrata_ultimate.governance.escalation_handler.notifications

Reference › orchestrata_ultimate › governance › escalation_handler › notifications

The notification channels for SLA breaches. NotificationChannel is a protocol with one method, async send(message) returning a success flag, so a channel for any medium can be supplied. LogNotificationChannel is the built-in implementation: it writes the message to a Python logger at warning level with the prefix SLA BREACH NOTIFICATION and always returns True. sla uses a list of channels, defaulting to one log channel, and the package initialiser re-exports both names.

At a glance

Class diagram

Dependency diagram

API

Classes

NotificationChannel (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 9

Protocol for SLA breach notification delivery.

LogNotificationChannel

Line 24

Notification channel that logs to Python logger.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/governance/escalation_handler/notifications.py

# mypy: ignore-errors
"""SLA breach notification channels."""

import logging
from typing import Protocol, runtime_checkable


@runtime_checkable
class NotificationChannel(Protocol):
    """Protocol for SLA breach notification delivery."""

    async def send(self, message: str) -> bool:
        """Send a notification message.

        Args:
            message: Notification message to send

        Returns:
            True if sent successfully, False otherwise
        """
        ...  # pragma: no cover


class LogNotificationChannel:
    """Notification channel that logs to Python logger."""

    def __init__(self, logger_name: str = __name__) -> None:
        """Initialize with a logger.

        Args:
            logger_name: Name of the logger to use
        """
        self._logger = logging.getLogger(logger_name)

    async def send(self, message: str) -> bool:
        """Send notification via logging.

        Args:
            message: Notification message to log

        Returns:
            True always (logging does not fail)
        """
        self._logger.warning(f"SLA BREACH NOTIFICATION: {message}")
        return True