Skip to content

sender — UI message delivery and synchronization

ViewSender

ViewSender

STATELESS service for sending and updating UI messages.

Works with two persistent bot messages in the chat:

  • Menu — navigation block (section buttons).
  • Content — information block (current feature data).

The send() algorithm:

  1. Deletes the trigger_message (e.g., the /start command).
  2. If clean_history=True — deletes old Menu and Content.
  3. Edits existing messages (or creates new ones).
  4. Saves current message_id via SenderManager.

.. note:: alert_text from UnifiedViewDTO is intentionally not processed by ViewSender — alert requires access to CallbackQuery.answer(), which ViewSender does not have. Call await call.answer(view.alert_text) in the handler before await sender.send(view).

Parameters:

Name Type Description Default
bot Bot

Aiogram Bot instance.

 required
manager SenderManager

SenderManager for storing UI coordinates.

 required
Example 
sender = ViewSender(bot=bot, manager=sender_manager)
# In a handler:
if view.alert_text:
    await call.answer(view.alert_text, show_alert=True)
await sender.send(view)
Source code in src/codex_bot/sender/view_sender.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class ViewSender:
    """STATELESS service for sending and updating UI messages.

    Works with two persistent bot messages in the chat:

    - **Menu** — navigation block (section buttons).
    - **Content** — information block (current feature data).

    The ``send()`` algorithm:

    1. Deletes the ``trigger_message`` (e.g., the ``/start`` command).
    2. If ``clean_history=True`` — deletes old Menu and Content.
    3. Edits existing messages (or creates new ones).
    4. Saves current ``message_id`` via SenderManager.

    .. note::
        ``alert_text`` from ``UnifiedViewDTO`` is intentionally not processed by ViewSender —
        alert requires access to ``CallbackQuery.answer()``, which ViewSender does not have.
        Call ``await call.answer(view.alert_text)`` in the handler before ``await sender.send(view)``.

    Args:
        bot: Aiogram Bot instance.
        manager: SenderManager for storing UI coordinates.

    Example:
        ```python
        sender = ViewSender(bot=bot, manager=sender_manager)
        # In a handler:
        if view.alert_text:
            await call.answer(view.alert_text, show_alert=True)
        await sender.send(view)
        ```
    """

    def __init__(self, bot: Bot, manager: SenderManager) -> None:
        self.bot = bot
        self.manager = manager

    async def send(self, view: UnifiedViewDTO) -> None:
        """Main UI synchronization method.

        All context variables are local, without writing to self.
        Safe for concurrent calls from different users.

        Args:
            view: ``UnifiedViewDTO`` from the orchestrator. Must contain
                  ``session_key`` and ``chat_id`` (filled by the Director).
        """
        if not view.session_key or not view.chat_id:
            log.error("ViewSender | missing session_key or chat_id in UnifiedViewDTO")
            return

        # Local variables — each send() call is isolated
        key = view.session_key
        chat_id = view.chat_id
        thread_id = view.message_thread_id
        is_channel = self._detect_channel(view)

        # 1. Delete trigger message (e.g., /start)
        if view.trigger_message_id:
            with contextlib.suppress(TelegramAPIError):
                await self.bot.delete_message(
                    chat_id=chat_id,
                    message_id=view.trigger_message_id,
                )

        coords = await self.manager.get_coords(key, is_channel)

        # 2. Clean old UI
        if view.clean_history:
            await self._delete_coords(coords, chat_id)
            coords = {}
            await self.manager.clear_coords(key, is_channel)

        # 3. Update Menu and Content (chat_id and thread_id explicitly via parameters)
        old_menu_id = coords.get("menu_msg_id")
        new_menu_id = await self._process_message(view.menu, old_menu_id, "MENU", chat_id, thread_id)

        old_content_id = coords.get("content_msg_id")
        new_content_id = await self._process_message(view.content, old_content_id, "CONTENT", chat_id, thread_id)

        # 4. Save updated coordinates
        updates: dict[str, int] = {}
        if new_menu_id and new_menu_id != old_menu_id:
            updates["menu_msg_id"] = new_menu_id
        if new_content_id and new_content_id != old_content_id:
            updates["content_msg_id"] = new_content_id

        if updates:
            await self.manager.update_coords(key, updates, is_channel)

    async def _delete_coords(
        self,
        coords: dict[str, int],
        chat_id: int | str,
    ) -> None:
        """Deletes Menu and Content messages from the chat.

        Args:
            coords: Dictionary with ``menu_msg_id`` and ``content_msg_id``.
            chat_id: Chat ID for deletion.
        """
        for msg_id in (coords.get("menu_msg_id"), coords.get("content_msg_id")):
            if msg_id:
                with contextlib.suppress(TelegramAPIError):
                    await self.bot.delete_message(chat_id=chat_id, message_id=msg_id)

    async def _process_message(
        self,
        view_dto: ViewResultDTO | None,
        old_message_id: int | None,
        log_prefix: str,
        chat_id: int | str,
        thread_id: int | None,
    ) -> int | None:
        """Edits an existing message or sends a new one.

        Args:
            view_dto: Content to send. None — skip.
            old_message_id: ID of the existing message to edit.
            log_prefix: Log label (``"MENU"`` or ``"CONTENT"``).
            chat_id: Target chat ID.
            thread_id: Topic ID in a supergroup.

        Returns:
            ID of the current message (old or new). None on error.
        """
        if not view_dto:
            return old_message_id

        if old_message_id:
            try:
                await self.bot.edit_message_text(
                    chat_id=chat_id,
                    message_id=old_message_id,
                    text=view_dto.text,
                    reply_markup=view_dto.kb,
                    parse_mode="HTML",
                )
                return old_message_id
            except TelegramBadRequest as e:
                if "message is not modified" in str(e).lower():
                    return old_message_id
                # Message deleted or inaccessible — create new
            except TelegramAPIError:
                pass

        try:
            sent = await self.bot.send_message(
                chat_id=chat_id,
                text=view_dto.text,
                reply_markup=view_dto.kb,
                message_thread_id=thread_id,
                parse_mode="HTML",
            )
            log.debug(f"ViewSender [{log_prefix}] sent | chat={chat_id} msg={sent.message_id}")
            return sent.message_id
        except TelegramAPIError as e:
            log.error(f"ViewSender [{log_prefix}] error | chat={chat_id} error='{e}'")
            return None

    @staticmethod
    def _detect_channel(view: UnifiedViewDTO) -> bool:
        """Detects if the chat is a channel or a group.

        Args:
            view: UnifiedViewDTO with chat data.

        Returns:
            ``True`` if it's a channel, topic, or group with a negative chat_id.
        """
        return (
            view.mode in ("channel", "topic")
            or (isinstance(view.chat_id, int) and view.chat_id < 0)
            or str(view.chat_id).startswith("-")
            or view.message_thread_id is not None
        )

Functions

send(view) async

Main UI synchronization method.

All context variables are local, without writing to self. Safe for concurrent calls from different users.

Parameters:

Name Type Description Default
view UnifiedViewDTO

UnifiedViewDTO from the orchestrator. Must contain session_key and chat_id (filled by the Director).

 required
Source code in src/codex_bot/sender/view_sender.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def send(self, view: UnifiedViewDTO) -> None:
    """Main UI synchronization method.

    All context variables are local, without writing to self.
    Safe for concurrent calls from different users.

    Args:
        view: ``UnifiedViewDTO`` from the orchestrator. Must contain
              ``session_key`` and ``chat_id`` (filled by the Director).
    """
    if not view.session_key or not view.chat_id:
        log.error("ViewSender | missing session_key or chat_id in UnifiedViewDTO")
        return

    # Local variables — each send() call is isolated
    key = view.session_key
    chat_id = view.chat_id
    thread_id = view.message_thread_id
    is_channel = self._detect_channel(view)

    # 1. Delete trigger message (e.g., /start)
    if view.trigger_message_id:
        with contextlib.suppress(TelegramAPIError):
            await self.bot.delete_message(
                chat_id=chat_id,
                message_id=view.trigger_message_id,
            )

    coords = await self.manager.get_coords(key, is_channel)

    # 2. Clean old UI
    if view.clean_history:
        await self._delete_coords(coords, chat_id)
        coords = {}
        await self.manager.clear_coords(key, is_channel)

    # 3. Update Menu and Content (chat_id and thread_id explicitly via parameters)
    old_menu_id = coords.get("menu_msg_id")
    new_menu_id = await self._process_message(view.menu, old_menu_id, "MENU", chat_id, thread_id)

    old_content_id = coords.get("content_msg_id")
    new_content_id = await self._process_message(view.content, old_content_id, "CONTENT", chat_id, thread_id)

    # 4. Save updated coordinates
    updates: dict[str, int] = {}
    if new_menu_id and new_menu_id != old_menu_id:
        updates["menu_msg_id"] = new_menu_id
    if new_content_id and new_content_id != old_content_id:
        updates["content_msg_id"] = new_content_id

    if updates:
        await self.manager.update_coords(key, updates, is_channel)

SenderManager

SenderManager

Manager for UI coordinate storage.

Abstracts work with coordinates (menu_msg_id, content_msg_id) on top of SenderStateStorageProtocol. ViewSender works only through it.

Parameters:

Name Type Description Default
storage SenderStateStorageProtocol

Implementation of SenderStateStorageProtocol (Redis, in-memory, etc.).

 required
Example 
manager = SenderManager(storage=redis_sender_storage)
coords = await manager.get_coords(session_key=123456)
# {"menu_msg_id": 10, "content_msg_id": 11}

await manager.update_coords(123456, {"menu_msg_id": 20})
await manager.clear_coords(123456)
Source code in src/codex_bot/sender/sender_manager.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class SenderManager:
    """
    Manager for UI coordinate storage.

    Abstracts work with coordinates (menu_msg_id, content_msg_id)
    on top of SenderStateStorageProtocol. ViewSender works only through it.

    Args:
        storage: Implementation of SenderStateStorageProtocol (Redis, in-memory, etc.).

    Example:
        ```python
        manager = SenderManager(storage=redis_sender_storage)
        coords = await manager.get_coords(session_key=123456)
        # {"menu_msg_id": 10, "content_msg_id": 11}

        await manager.update_coords(123456, {"menu_msg_id": 20})
        await manager.clear_coords(123456)
        ```
    """

    def __init__(self, storage: SenderStateStorageProtocol) -> None:
        self.storage = storage

    async def get_coords(self, session_key: int | str, is_channel: bool = False) -> dict[str, int]:
        """
        Returns saved UI coordinates for a session.

        Args:
            session_key: user_id (int) or session_id (str) for a channel.
            is_channel: True — use channel key, False — use user key.

        Returns:
            Dictionary {"menu_msg_id": int, "content_msg_id": int}.
            Empty dictionary if no data exists.
        """
        key = self._build_key(session_key, is_channel)
        return await self.storage.get_sender_state(key)

    async def update_coords(
        self,
        session_key: int | str,
        coords: dict[str, int],
        is_channel: bool = False,
    ) -> None:
        """
        Partially updates UI coordinates for a session.

        Args:
            session_key: user_id or session_id.
            coords: Fields to update (partial update).
            is_channel: Key type.
        """
        if not coords:
            return
        key = self._build_key(session_key, is_channel)
        await self.storage.save_sender_state(key, coords)

    async def clear_coords(self, session_key: int | str, is_channel: bool = False) -> None:
        """
        Deletes all UI coordinates for a session (state reset).

        Args:
            session_key: user_id or session_id.
            is_channel: Key type.
        """
        key = self._build_key(session_key, is_channel)
        await self.storage.clear_sender_state(key)

    @staticmethod
    def _build_key(session_key: int | str, is_channel: bool) -> str:
        if is_channel:
            return SenderKeys.channel(str(session_key))
        return SenderKeys.user(session_key)

Functions

clear_coords(session_key, is_channel=False) async

Deletes all UI coordinates for a session (state reset).

Parameters:

Name Type Description Default
session_key int | str

user_id or session_id.

 required
is_channel bool

Key type.

 False
Source code in src/codex_bot/sender/sender_manager.py 
70
71
72
73
74
75
76
77
78
79
async def clear_coords(self, session_key: int | str, is_channel: bool = False) -> None:
    """
    Deletes all UI coordinates for a session (state reset).

    Args:
        session_key: user_id or session_id.
        is_channel: Key type.
    """
    key = self._build_key(session_key, is_channel)
    await self.storage.clear_sender_state(key)

get_coords(session_key, is_channel=False) async

Returns saved UI coordinates for a session.

Parameters:

Name Type Description Default
session_key int | str

user_id (int) or session_id (str) for a channel.

 required
is_channel bool

True — use channel key, False — use user key.

 False

Returns:

Type Description
dict[str, int]

Dictionary {"menu_msg_id": int, "content_msg_id": int}.
dict[str, int]

Empty dictionary if no data exists.
Source code in src/codex_bot/sender/sender_manager.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
async def get_coords(self, session_key: int | str, is_channel: bool = False) -> dict[str, int]:
    """
    Returns saved UI coordinates for a session.

    Args:
        session_key: user_id (int) or session_id (str) for a channel.
        is_channel: True — use channel key, False — use user key.

    Returns:
        Dictionary {"menu_msg_id": int, "content_msg_id": int}.
        Empty dictionary if no data exists.
    """
    key = self._build_key(session_key, is_channel)
    return await self.storage.get_sender_state(key)

update_coords(session_key, coords, is_channel=False) async

Partially updates UI coordinates for a session.

Parameters:

Name Type Description Default
session_key int | str

user_id or session_id.

 required
coords dict[str, int]

Fields to update (partial update).

 required
is_channel bool

Key type.

 False
Source code in src/codex_bot/sender/sender_manager.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def update_coords(
    self,
    session_key: int | str,
    coords: dict[str, int],
    is_channel: bool = False,
) -> None:
    """
    Partially updates UI coordinates for a session.

    Args:
        session_key: user_id or session_id.
        coords: Fields to update (partial update).
        is_channel: Key type.
    """
    if not coords:
        return
    key = self._build_key(session_key, is_channel)
    await self.storage.save_sender_state(key, coords)

SenderKeys

SenderKeys

Key factory for SenderManager.

Standardizes key naming in the UI coordinate storage. Use in your implementation of SenderStateStorageProtocol.

Example 
key = SenderKeys.user(user_id=123456789)
# "sender:user:123456789"

key = SenderKeys.channel(session_id="booking_feed_1")
# "sender:channel:booking_feed_1"
Source code in src/codex_bot/sender/sender_keys.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class SenderKeys:
    """
    Key factory for SenderManager.

    Standardizes key naming in the UI coordinate storage.
    Use in your implementation of SenderStateStorageProtocol.

    Example:
        ```python
        key = SenderKeys.user(user_id=123456789)
        # "sender:user:123456789"

        key = SenderKeys.channel(session_id="booking_feed_1")
        # "sender:channel:booking_feed_1"
        ```
    """

    @staticmethod
    def user(user_id: int | str) -> str:
        """
        Key for storing UI coordinates of private correspondence.

        Args:
            user_id: Telegram ID of the user.

        Returns:
            Key string like `sender:user:<user_id>`.
        """
        return f"sender:user:{user_id}"

    @staticmethod
    def channel(session_id: str) -> str:
        """
        Key for storing UI coordinates of a channel or group.

        Args:
            session_id: Unique identifier of the channel session.

        Returns:
            Key string like `sender:channel:<session_id>`.
        """
        return f"sender:channel:{session_id}"

Functions

channel(session_id) staticmethod

Key for storing UI coordinates of a channel or group.

Parameters:

Name Type Description Default
session_id str

Unique identifier of the channel session.

 required

Returns:

Type Description
str

Key string like sender:channel:<session_id>.
Source code in src/codex_bot/sender/sender_keys.py 
39
40
41
42
43
44
45
46
47
48
49
50
@staticmethod
def channel(session_id: str) -> str:
    """
    Key for storing UI coordinates of a channel or group.

    Args:
        session_id: Unique identifier of the channel session.

    Returns:
        Key string like `sender:channel:<session_id>`.
    """
    return f"sender:channel:{session_id}"

user(user_id) staticmethod

Key for storing UI coordinates of private correspondence.

Parameters:

Name Type Description Default
user_id int | str

Telegram ID of the user.

 required

Returns:

Type Description
str

Key string like sender:user:<user_id>.
Source code in src/codex_bot/sender/sender_keys.py 
26
27
28
29
30
31
32
33
34
35
36
37
@staticmethod
def user(user_id: int | str) -> str:
    """
    Key for storing UI coordinates of private correspondence.

    Args:
        user_id: Telegram ID of the user.

    Returns:
        Key string like `sender:user:<user_id>`.
    """
    return f"sender:user:{user_id}"

Protocols

SenderStateStorageProtocol

Bases: Protocol

Contract for UI coordinate storage (Menu and Content message IDs).

Implement this protocol in the project via Redis, PostgreSQL, or in-memory — ViewSender and SenderManager do not know about the specific storage.

Example 
class RedisSenderStorage:
    def __init__(self, redis: Redis): ...

    async def get_sender_state(self, key: str) -> dict[str, int]:
        data = await redis.hgetall(f"sender:{key}")
        return {k: int(v) for k, v in data.items()}

    async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
        await redis.hset(f"sender:{key}", mapping={k: str(v) for k, v in data.items()})

    async def clear_sender_state(self, key: str) -> None:
        await redis.delete(f"sender:{key}")
Source code in src/codex_bot/sender/protocols.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@runtime_checkable
class SenderStateStorageProtocol(Protocol):
    """
    Contract for UI coordinate storage (Menu and Content message IDs).

    Implement this protocol in the project via Redis, PostgreSQL, or in-memory —
    ViewSender and SenderManager do not know about the specific storage.

    Example:
        ```python
        class RedisSenderStorage:
            def __init__(self, redis: Redis): ...

            async def get_sender_state(self, key: str) -> dict[str, int]:
                data = await redis.hgetall(f"sender:{key}")
                return {k: int(v) for k, v in data.items()}

            async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
                await redis.hset(f"sender:{key}", mapping={k: str(v) for k, v in data.items()})

            async def clear_sender_state(self, key: str) -> None:
                await redis.delete(f"sender:{key}")
        ```
    """

    async def get_sender_state(self, key: str) -> dict[str, int]:
        """
        Returns UI coordinates (message IDs) for a session.

        Args:
            key: Session key (e.g., user_id string or channel session_id).

        Returns:
            Dictionary like {"menu_msg_id": 123, "content_msg_id": 124}.
            Empty dictionary if no data exists.
        """
        ...

    async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
        """
        Saves (partially updates) UI coordinates for a session.

        Args:
            key: Session key.
            data: Fields to update, e.g., {"menu_msg_id": 123}.
        """
        ...

    async def clear_sender_state(self, key: str) -> None:
        """
        Deletes all UI coordinates for a session.

        Args:
            key: Session key.
        """
        ...

Functions

clear_sender_state(key) async

Deletes all UI coordinates for a session.

Parameters:

Name Type Description Default
key str

Session key.

 required
Source code in src/codex_bot/sender/protocols.py 
60
61
62
63
64
65
66
67
async def clear_sender_state(self, key: str) -> None:
    """
    Deletes all UI coordinates for a session.

    Args:
        key: Session key.
    """
    ...

get_sender_state(key) async

Returns UI coordinates (message IDs) for a session.

Parameters:

Name Type Description Default
key str

Session key (e.g., user_id string or channel session_id).

 required

Returns:

Type Description
dict[str, int]

Dictionary like {"menu_msg_id": 123, "content_msg_id": 124}.
dict[str, int]

Empty dictionary if no data exists.
Source code in src/codex_bot/sender/protocols.py 
37
38
39
40
41
42
43
44
45
46
47
48
async def get_sender_state(self, key: str) -> dict[str, int]:
    """
    Returns UI coordinates (message IDs) for a session.

    Args:
        key: Session key (e.g., user_id string or channel session_id).

    Returns:
        Dictionary like {"menu_msg_id": 123, "content_msg_id": 124}.
        Empty dictionary if no data exists.
    """
    ...

save_sender_state(key, data) async

Saves (partially updates) UI coordinates for a session.

Parameters:

Name Type Description Default
key str

Session key.

 required
data dict[str, int]

Fields to update, e.g., {"menu_msg_id": 123}.

 required
Source code in src/codex_bot/sender/protocols.py 
50
51
52
53
54
55
56
57
58
async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
    """
    Saves (partially updates) UI coordinates for a session.

    Args:
        key: Session key.
        data: Fields to update, e.g., {"menu_msg_id": 123}.
    """
    ...