TTS API Notifier

A microservice that consumes alarm notifications from RabbitMQ and creates/updates tickets in a Trouble Ticket System (TTS) via an abstract TTS service interface.

Architecture

The service runs two threads:

Main thread: Reads messages from RabbitMQ and enqueues valid messages
Processing thread (daemon): Processes messages from queue with retry logic

Message Flow

Components

Modules

__init__.py - Main entry point with start_service() function
service.py - Core processing logic and retry mechanism
tts_interface.py - Abstract TTS service interface (TTSService)
config.py - Configuration loading and validation
exceptions.py - Custom exceptions (TicketError)
cli.py - CLI with dynamic service loading

TTS Service Interface

The TTSService abstract base class defines three methods:

from datetime import datetime
from dataclasses import dataclass

@dataclass
class TTSTicket:
    ticket_id: str        # Internal ticket ID used for updates
    ticket_number: str    # Display ticket number (e.g., "MAINT-123")
    comment: str          # Description or comment about the ticket
    ticket_link: str      # URL to view the ticket in the TTS

class TTSService(ABC):
    @abstractmethod
    def find_overlapping_ticket(self, alarm_time: datetime, services: set[str]) -> TTSTicket | None:
        """
        Return TTSTicket object of overlapping maintenance ticket, or None.

        Args:
            alarm_time: The alarm timestamp (datetime) to check
            services: Set of service names/identifiers affected by the alarm

        Returns:
            TTSTicket or None
        """
        pass

    @abstractmethod
    def update_maintenance_ticket(self, ticket_id: str, message: dict) -> None:
        """Update existing TTS ticket with message details."""
        pass

    @abstractmethod
    def create_ticket(self, message: dict) -> str:
        """Create new TTS ticket and return its ticket_id."""
        pass