INNER CODE UNIT · Python
TelegramNotificationService
PleasePrompto/ductor · ductor_bot/messenger/telegram/app.py:158
class TelegramNotificationService:
"""NotificationService implementation for Telegram."""
def __init__(self, bot: Bot, config: AgentConfig) -> None:
self._bot = bot
self._config = config
async def notify(self, chat_id: int, text: str) -> None:
await send_rich(self._bot, chat_id, text, None)
async def notify_all(self, text: str) -> None:
for uid in self._config.allowed_user_ids:
try:
await send_rich(self._bot, uid, text, None)
except TelegramAPIError:
# An unreachable recipient (e.g. never pressed /start -> "chat
# not found") must not abort the notification fan-out — or, at
# startup, the whole boot.