INNER CODE UNIT · Python

_cancel_task

PleasePrompto/ductor · ductor_bot/messenger/telegram/app.py:150

async def _cancel_task(task: asyncio.Task[None] | None) -> None:
    """Cancel an asyncio task and suppress CancelledError."""
    if task and not task.done():
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task


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)

View source record →