connector.py

Go to the documentation of this file.

"""Telegram bundle adapter used by the durable outbox.
 
Dependencies
------------
- oc_connect.connectors.telegram.destinations (DestinationStore - resolves the logical destination to a chat and credential name)
- oc_connect.connectors.telegram.transport (TelegramTransport - performs the raw Bot API send)
- oc_connect.credentials (CredentialBroker - retrieves the bot token by name)
- oc_connect.outbox (OutboxItem - the already-filtered item being delivered)
"""
from __future__ import annotations
 
from collections.abc import Mapping
from typing import Any
 
from ...credentials import CredentialBroker
from ...outbox import OutboxItem
from .destinations import DestinationStore
from .transport import TelegramTransport
 
 
class TelegramConnector:
    """Resolve one logical destination and perform one raw Bot API send."""
 
    def __init__(
        self,
        broker: CredentialBroker,
        destinations: DestinationStore,
    ) -> None:
        self.broker = broker
        self.destinations = destinations
 
    def deliver(
        self,
        item: OutboxItem,
        *,
        reply_markup: Mapping[str, Any] | None = None,
    ) -> str | None:
        """Deliver an already-filtered outbox item, returning its message id.
 
        The returned ``message_id`` is what an operator reply's ``reply_to``
        metadata resolves against when the inbound poller correlates a
        structured reply to the question it answers (§4). *reply_markup*
        (plan 236) is the already-built vocabulary keyboard, or ``None`` for a
        plain send; the item's content is unchanged either way.
        """
        destination = self.destinations.get(item.destination)
        raw_value = self.broker.retrieve(destination.credential_name)
        transport = TelegramTransport(raw_value)
        try:
            result = transport.send_message(
                destination.chat_id, item.content, reply_markup=reply_markup
            )
        finally:
            transport.close()
        message_id = result.get("message_id")
        return str(message_id) if message_id is not None else None
 
 
__all__ = ["TelegramConnector"]