wrongmove/crawler/notifications.py

from abc import abstractmethod
import apprise
from functools import lru_cache
import os


class Surface:
    @abstractmethod
    def connection_string(self) -> str | None: ...


class Slack(Surface):
    def connection_string(self) -> str | None:
        return os.environ.get("SLACK_WEBHOOK_URL")


@lru_cache(maxsize=None)
def get_notifier(surfaces: list[Surface] | None = None) -> apprise.Apprise:
    surfaces = surfaces or list[Surface]([Slack()])
    obj = apprise.Apprise()
    for surface in surfaces:
        if conn := surface.connection_string():
            obj.add(conn)
    return obj


async def send_notification(body: str, title: str = "") -> bool:
    notifier = get_notifier()
    return await notifier.async_notify(body=body, title=title)