notifiers.py

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Callable, Optional, Protocol, Type, TYPE_CHECKING
import httpx

from yarl import URL

from intent_deployer.config import get_settings, telegram_url

log: logging.Logger = logging.getLogger(__name__)

if TYPE_CHECKING:
    # BUG: in py3.9.7 data classes that inherit from typing.Protocol break so
    # only actually use Protocol for type checking, use a plain class otherwise
    class Notifier(Protocol):
        @classmethod
        def parse(cls, body: dict) -> Notifier:
            ...

        async def send(self, msg: str):
            ...
else:
    class Notifier:
        pass


_notifiers: dict[str, Type[Notifier]] = {}


def register_notifier(name: str) -> Callable[[Type[Notifier]], Type[Notifier]]:
    if name in _notifiers:
        log.warning("The notifier type %s has already been registered", name)

    def wrapper(notifier: Type[Notifier]) -> Type[Notifier]:
        _notifiers[name] = notifier

        return notifier

    return wrapper


def get_notifier(name: str, body: dict) -> Optional[Notifier]:
    klass = _notifiers.get(name)
    if klass is None:
        return None
    return klass.parse(body)


@register_notifier("telegram")
@dataclass
class Telegram(Notifier):
    url: URL

    @classmethod
    def parse(cls, body: dict) -> Telegram:
        chat_id = body["data"]["chat"]["id"]
        token = get_settings().telegram_token
        if token is None:
            raise Exception(
                "Tried to create a telegram notifier, but no TELEGRAM_TOKEN was provided"
            )

        url = telegram_url / f"bot{token}" / "sendMessage" % {"chat_id": chat_id}

        return cls(url)

    async def send(self, msg: str):
        url = self.url % {"text": msg}
        async with httpx.AsyncClient() as client:
            await client.post(str(url))


class NullNotifier(Notifier):
    @classmethod
    def parse(cls, body: dict) -> NullNotifier:
        return cls()

    async def send(self, msg: str):
        pass