sender — UI message delivery and synchronization

Layer for managing display and synchronization of messages in Telegram.

ViewSender

ViewSender

Stateless service for persistent UI orchestration.

The ViewSender implements a "Dual-Message" UI pattern, maintaining two distinct messages per session:

1. Menu: A high-level navigation block for switching features.
2. Content: A dynamic block representing the current feature's state.

The synchronization algorithm performs atomic updates to these blocks, handling message deletion (e.g., of trigger commands) and state persistence via an external SenderManager.

Note

alert_text in UnifiedViewDTO is not handled by this service as it requires a CallbackQuery context. It must be answered manually in the handler prior to calling send().

Parameters:

Name	Type	Description	Default
bot	Bot	Instance of the aiogram.Bot used for Telegram API interactions.	required
manager	SenderManager	A SenderManager providing access to UI coordinate persistence.	required

Source code in src/codex_bot/sender/view_sender.py

class ViewSender:
    """Stateless service for persistent UI orchestration.

    The ViewSender implements a "Dual-Message" UI pattern, maintaining two
    distinct messages per session:

    1. **Menu**: A high-level navigation block for switching features.
    2. **Content**: A dynamic block representing the current feature's state.

    The synchronization algorithm performs atomic updates to these blocks,
    handling message deletion (e.g., of trigger commands) and state
    persistence via an external `SenderManager`.

    Note:
        `alert_text` in `UnifiedViewDTO` is not handled by this service as it
        requires a `CallbackQuery` context. It must be answered manually in
        the handler prior to calling `send()`.

    Args:
        bot: Instance of the `aiogram.Bot` used for Telegram API interactions.
        manager: A `SenderManager` providing access to UI coordinate persistence.
    """

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

    async def send(self, view: UnifiedViewDTO) -> None:
        """Synchronize the current UI state with Telegram.

        Analyzes the `UnifiedViewDTO` to perform required deletions, updates,
        and message deliveries. This method is thread-safe and stateless,
        operating strictly on the provided DTO.

        Args:
            view: The structured UI definition from the orchestrator.
                Must contain valid routing metadata (chat_id, session_key).

        Side Effects:
            - Deletes the trigger message if `trigger_message_id` is present.
            - Updates persistent UI coordinates in the backend storage.
            - Modifies existing messages in the target Telegram chat.
        """
        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
        chat_id = view.chat_id
        thread_id = view.message_thread_id
        is_channel = self._detect_channel(view)

        # Logic: In public chats, UI messages are shared.
        # We ignore session_key and bind coordinates to chat/topic ID.
        key = (f"{chat_id}:{thread_id}" if thread_id else str(chat_id)) if is_channel else str(view.session_key)

        # 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 safely.

        Args:
            coords: Dictionary with ``menu_msg_id`` and ``content_msg_id``.
            chat_id: Chat ID for deletion.
        """
        # Use set to avoid double deletion of the same ID and filter None
        msg_ids = set(filter(None, (coords.get("menu_msg_id"), coords.get("content_msg_id"))))

        for msg_id in msg_ids:
            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

Synchronize the current UI state with Telegram.

Analyzes the UnifiedViewDTO to perform required deletions, updates, and message deliveries. This method is thread-safe and stateless, operating strictly on the provided DTO.

Parameters:

Name	Type	Description	Default
view	UnifiedViewDTO	The structured UI definition from the orchestrator. Must contain valid routing metadata (chat_id, session_key).	required

Side Effects

• Deletes the trigger message if trigger_message_id is present.
• Updates persistent UI coordinates in the backend storage.
• Modifies existing messages in the target Telegram chat.

Source code in src/codex_bot/sender/view_sender.py

async def send(self, view: UnifiedViewDTO) -> None:
    """Synchronize the current UI state with Telegram.

    Analyzes the `UnifiedViewDTO` to perform required deletions, updates,
    and message deliveries. This method is thread-safe and stateless,
    operating strictly on the provided DTO.

    Args:
        view: The structured UI definition from the orchestrator.
            Must contain valid routing metadata (chat_id, session_key).

    Side Effects:
        - Deletes the trigger message if `trigger_message_id` is present.
        - Updates persistent UI coordinates in the backend storage.
        - Modifies existing messages in the target Telegram chat.
    """
    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
    chat_id = view.chat_id
    thread_id = view.message_thread_id
    is_channel = self._detect_channel(view)

    # Logic: In public chats, UI messages are shared.
    # We ignore session_key and bind coordinates to chat/topic ID.
    key = (f"{chat_id}:{thread_id}" if thread_id else str(chat_id)) if is_channel else str(view.session_key)

    # 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

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

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

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

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)

---

Storage Implementations

Ready-to-use implementations for storing UI coordinates.

MemorySenderStorage

MemorySenderStorage

Bases: SenderStateStorageProtocol

In-memory storage for UI coordinates.

Ideal for lightweight bots or development environments where Redis is not available. Data is stored in a local dictionary and persists only for the duration of the bot's process life.

Example

if not settings.use_redis:
    storage = MemorySenderStorage()
    manager = SenderManager(storage=storage)
    view_sender = ViewSender(bot=bot, manager=manager)

Source code in src/codex_bot/sender/memory_storage.py

class MemorySenderStorage(SenderStateStorageProtocol):
    """In-memory storage for UI coordinates.

    Ideal for lightweight bots or development environments where Redis is not available.
    Data is stored in a local dictionary and persists only for the duration
    of the bot's process life.

    Example:
        ```python
        if not settings.use_redis:
            storage = MemorySenderStorage()
            manager = SenderManager(storage=storage)
            view_sender = ViewSender(bot=bot, manager=manager)
        ```
    """

    def __init__(self) -> None:
        """Initializes the local storage dictionary."""
        self._storage: dict[str, dict[str, int]] = {}

    async def get_sender_state(self, key: str) -> dict[str, int]:
        """Retrieves coordinates from memory.

        Args:
            key: Storage key (e.g., "sender:user:123").

        Returns:
            A copy of the coordinate dictionary. Returns an empty dict if not found.
        """
        data = self._storage.get(key, {})
        # Return a copy to prevent accidental mutations by reference
        return data.copy()

    async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
        """Saves or updates coordinates in memory.

        Args:
            key: Storage key.
            data: Dictionary with updated coordinates (e.g., {"menu_msg_id": 42}).
        """
        if key not in self._storage:
            self._storage[key] = {}

        self._storage[key].update(data)

    async def clear_sender_state(self, key: str) -> None:
        """Removes coordinates for the specified key from memory.

        Args:
            key: Storage key to clear.
        """
        self._storage.pop(key, None)

Functions

__init__()

Initializes the local storage dictionary.

Source code in src/codex_bot/sender/memory_storage.py

def __init__(self) -> None:
    """Initializes the local storage dictionary."""
    self._storage: dict[str, dict[str, int]] = {}

clear_sender_state(key) async

Removes coordinates for the specified key from memory.

Parameters:

Name	Type	Description	Default
key	str	Storage key to clear.	required

Source code in src/codex_bot/sender/memory_storage.py

async def clear_sender_state(self, key: str) -> None:
    """Removes coordinates for the specified key from memory.

    Args:
        key: Storage key to clear.
    """
    self._storage.pop(key, None)

get_sender_state(key) async

Retrieves coordinates from memory.

Parameters:

Name	Type	Description	Default
key	str	Storage key (e.g., "sender:user:123").	required

Returns:

Type	Description
dict[str, int]	A copy of the coordinate dictionary. Returns an empty dict if not found.

Source code in src/codex_bot/sender/memory_storage.py

async def get_sender_state(self, key: str) -> dict[str, int]:
    """Retrieves coordinates from memory.

    Args:
        key: Storage key (e.g., "sender:user:123").

    Returns:
        A copy of the coordinate dictionary. Returns an empty dict if not found.
    """
    data = self._storage.get(key, {})
    # Return a copy to prevent accidental mutations by reference
    return data.copy()

save_sender_state(key, data) async

Saves or updates coordinates in memory.

Parameters:

Name	Type	Description	Default
key	str	Storage key.	required
data	dict[str, int]	Dictionary with updated coordinates (e.g., {"menu_msg_id": 42}).	required

Source code in src/codex_bot/sender/memory_storage.py

async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
    """Saves or updates coordinates in memory.

    Args:
        key: Storage key.
        data: Dictionary with updated coordinates (e.g., {"menu_msg_id": 42}).
    """
    if key not in self._storage:
        self._storage[key] = {}

    self._storage[key].update(data)

RedisSenderStorage

RedisSenderStorage

Bases: SenderStateStorageProtocol

Redis storage implementation using HASHES.

Coordinates are stored as fields within a Redis Hash, allowing partial updates (e.g. only updating menu_msg_id) without rewriting the entire record.

Parameters:

Name	Type	Description	Default
redis	Any	An initialized Redis client (redis.asyncio).	required
ttl	int	Time-to-live for coordinate keys in seconds (default: 7 days).	604800

Source code in src/codex_bot/sender/redis_storage.py

class RedisSenderStorage(SenderStateStorageProtocol):
    """Redis storage implementation using HASHES.

    Coordinates are stored as fields within a Redis Hash, allowing
    partial updates (e.g. only updating menu_msg_id) without rewriting
    the entire record.

    Args:
        redis: An initialized Redis client (redis.asyncio).
        ttl: Time-to-live for coordinate keys in seconds (default: 7 days).
    """

    def __init__(self, redis: Any, ttl: int = 604800) -> None:
        self.redis = redis
        self.ttl = ttl

    async def get_sender_state(self, key: str) -> dict[str, int]:
        """Retrieves coordinates using HGETALL.

        Args:
            key: Prepared Redis key from SenderManager.

        Returns:
            Dictionary with coordinates. Keys and values are cast to correct types.
        """
        data = await self.redis.hgetall(key)
        if not data:
            return {}

        # Redis returns strings/bytes, we cast them back to integers
        return {k: int(v) for k, v in data.items()}

    async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
        """Saves or updates coordinates using HSET.

        Args:
            key: Prepared Redis key.
            data: Fields to update (e.g. {"menu_msg_id": 123}).
        """
        if not data:
            return

        # Atomic update of multiple fields in the hash
        await self.redis.hset(name=key, mapping=data)

        # Refresh TTL on every update
        await self.redis.expire(name=key, time=self.ttl)

    async def clear_sender_state(self, key: str) -> None:
        """Removes the entire hash key from Redis.

        Args:
            key: Prepared Redis key.
        """
        await self.redis.delete(key)

Functions

clear_sender_state(key) async

Removes the entire hash key from Redis.

Parameters:

Name	Type	Description	Default
key	str	Prepared Redis key.	required

Source code in src/codex_bot/sender/redis_storage.py

async def clear_sender_state(self, key: str) -> None:
    """Removes the entire hash key from Redis.

    Args:
        key: Prepared Redis key.
    """
    await self.redis.delete(key)

get_sender_state(key) async

Retrieves coordinates using HGETALL.

Parameters:

Name	Type	Description	Default
key	str	Prepared Redis key from SenderManager.	required

Returns:

Type	Description
dict[str, int]	Dictionary with coordinates. Keys and values are cast to correct types.

Source code in src/codex_bot/sender/redis_storage.py

async def get_sender_state(self, key: str) -> dict[str, int]:
    """Retrieves coordinates using HGETALL.

    Args:
        key: Prepared Redis key from SenderManager.

    Returns:
        Dictionary with coordinates. Keys and values are cast to correct types.
    """
    data = await self.redis.hgetall(key)
    if not data:
        return {}

    # Redis returns strings/bytes, we cast them back to integers
    return {k: int(v) for k, v in data.items()}

save_sender_state(key, data) async

Saves or updates coordinates using HSET.

Parameters:

Name	Type	Description	Default
key	str	Prepared Redis key.	required
data	dict[str, int]	Fields to update (e.g. {"menu_msg_id": 123}).	required

Source code in src/codex_bot/sender/redis_storage.py

async def save_sender_state(self, key: str, data: dict[str, int]) -> None:
    """Saves or updates coordinates using HSET.

    Args:
        key: Prepared Redis key.
        data: Fields to update (e.g. {"menu_msg_id": 123}).
    """
    if not data:
        return

    # Atomic update of multiple fields in the hash
    await self.redis.hset(name=key, mapping=data)

    # Refresh TTL on every update
    await self.redis.expire(name=key, time=self.ttl)

---

Infrastructure Utils

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

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

@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

@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

@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

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

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

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}.
    """
    ...