Skip to content

helper — Utility helpers

Module of auxiliary tools to simplify development and debugging.

ContextHelper

Universal extractor of identifiers from any type of events (Telegram or Redis).

ContextHelper

Utility class for polymorphic context extraction.

Standardizes the retrieval of identity and routing information across different interaction models. It ensures that business logic receives a consistent BaseBotContext regardless of the upstream event source.

Capabilities
  • Parsing aiogram.Message and aiogram.CallbackQuery.
  • Normalizing raw dictionaries (Redis Stream / Direct API).
  • Handling fallback logic for user/chat identity resolution.
Source code in src/codex_bot/helper/context_helper.py 
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
class ContextHelper:
    """Utility class for polymorphic context extraction.

    Standardizes the retrieval of identity and routing information across
    different interaction models. It ensures that business logic receives
    a consistent `BaseBotContext` regardless of the upstream event source.

    Capabilities:
        - Parsing `aiogram.Message` and `aiogram.CallbackQuery`.
        - Normalizing raw dictionaries (Redis Stream / Direct API).
        - Handling fallback logic for user/chat identity resolution.
    """

    @staticmethod
    def extract_base_context(event: TelegramObject | dict[str, Any]) -> BaseBotContext:
        """Analyze an event source and extract its routing metadata.

        Coordinates the extraction of User ID, Chat ID, and Thread IDs,
        applying normalization rules to ensure session consistency.

        Args:
            event: A native Telegram object or a dictionary representing
                an incoming event payload.

        Returns:
            An immutable `BaseBotContext` populated with the extracted IDs.
        """
        user_id: int = 0
        chat_id: int = 0
        message_id: int | None = None
        thread_id: int | None = None

        # 1. Handling aiogram objects
        if isinstance(event, Message):
            user_id = event.from_user.id if event.from_user else 0
            chat_id = event.chat.id
            message_id = event.message_id
            thread_id = event.message_thread_id

        elif isinstance(event, CallbackQuery):
            user_id = event.from_user.id
            if event.message and isinstance(event.message, Message):
                chat_id = event.message.chat.id
                message_id = event.message.message_id
                thread_id = event.message.message_thread_id
            else:
                chat_id = user_id

        # 2. Handling raw dictionaries (Redis / API)
        elif isinstance(event, dict):
            user_id = event.get("user_id") or event.get("from_id") or 0
            chat_id = event.get("chat_id") or user_id
            message_id = event.get("message_id") or event.get("trigger_id")
            thread_id = event.get("thread_id") or event.get("message_thread_id")

        # 3. Final normalization
        # Fallback: if user_id is missing but we have chat_id, use chat_id as session key
        if user_id == 0 and chat_id != 0:
            user_id = chat_id

        # If we still have no IDs, return default zeros (to avoid None errors in Director)
        return BaseBotContext(
            user_id=int(user_id),
            chat_id=int(chat_id),
            message_id=int(message_id) if message_id else None,
            message_thread_id=int(thread_id) if thread_id else None,
        )

Functions

extract_base_context(event) staticmethod

Analyze an event source and extract its routing metadata.

Coordinates the extraction of User ID, Chat ID, and Thread IDs, applying normalization rules to ensure session consistency.

Parameters:

Name Type Description Default
event TelegramObject | dict[str, Any]

A native Telegram object or a dictionary representing an incoming event payload.

 required

Returns:

Type Description
BaseBotContext

An immutable BaseBotContext populated with the extracted IDs.
Source code in src/codex_bot/helper/context_helper.py 
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
@staticmethod
def extract_base_context(event: TelegramObject | dict[str, Any]) -> BaseBotContext:
    """Analyze an event source and extract its routing metadata.

    Coordinates the extraction of User ID, Chat ID, and Thread IDs,
    applying normalization rules to ensure session consistency.

    Args:
        event: A native Telegram object or a dictionary representing
            an incoming event payload.

    Returns:
        An immutable `BaseBotContext` populated with the extracted IDs.
    """
    user_id: int = 0
    chat_id: int = 0
    message_id: int | None = None
    thread_id: int | None = None

    # 1. Handling aiogram objects
    if isinstance(event, Message):
        user_id = event.from_user.id if event.from_user else 0
        chat_id = event.chat.id
        message_id = event.message_id
        thread_id = event.message_thread_id

    elif isinstance(event, CallbackQuery):
        user_id = event.from_user.id
        if event.message and isinstance(event.message, Message):
            chat_id = event.message.chat.id
            message_id = event.message.message_id
            thread_id = event.message.message_thread_id
        else:
            chat_id = user_id

    # 2. Handling raw dictionaries (Redis / API)
    elif isinstance(event, dict):
        user_id = event.get("user_id") or event.get("from_id") or 0
        chat_id = event.get("chat_id") or user_id
        message_id = event.get("message_id") or event.get("trigger_id")
        thread_id = event.get("thread_id") or event.get("message_thread_id")

    # 3. Final normalization
    # Fallback: if user_id is missing but we have chat_id, use chat_id as session key
    if user_id == 0 and chat_id != 0:
        user_id = chat_id

    # If we still have no IDs, return default zeros (to avoid None errors in Director)
    return BaseBotContext(
        user_id=int(user_id),
        chat_id=int(chat_id),
        message_id=int(message_id) if message_id else None,
        message_thread_id=int(thread_id) if thread_id else None,
    )

ID Inspector

Ready-to-use handler for obtaining technical information about chat and user.

inspect_ids_handler(message) async

Ready-to-use aiogram handler for inspecting IDs.

Returns User ID, Chat ID, Chat Type, and Thread ID. Supports forwarded messages from channels to find Channel IDs.

Usage 
from codex_bot.helper.id_inspector import inspect_ids_handler
router.message(Command("id"))(inspect_ids_handler)
Source code in src/codex_bot/helper/id_inspector.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
async def inspect_ids_handler(message: Message) -> None:
    """
    Ready-to-use aiogram handler for inspecting IDs.

    Returns User ID, Chat ID, Chat Type, and Thread ID.
    Supports forwarded messages from channels to find Channel IDs.

    Usage:
        ```python
        from codex_bot.helper.id_inspector import inspect_ids_handler
        router.message(Command("id"))(inspect_ids_handler)
        ```
    """
    chat_type = message.chat.type
    user_id = message.from_user.id if message.from_user else "Unknown"

    report = (
        "🆔 <b>ID Inspector</b>\n"
        "──────────────────\n"
        f"👤 <b>User ID:</b> <code>{user_id}</code>\n"
        f"💬 <b>Chat ID:</b> <code>{message.chat.id}</code>\n"
        f"🏗 <b>Chat Type:</b> <code>{chat_type}</code>\n"
    )

    if message.message_thread_id:
        report += f"🧵 <b>Thread ID:</b> <code>{message.message_thread_id}</code>\n"

    if message.reply_to_message and message.reply_to_message.forward_from_chat:
        f_chat = message.reply_to_message.forward_from_chat
        report += f"📢 <b>Forwarded Chat ID:</b> <code>{f_chat.id}</code>\n"

    await message.reply(report, parse_mode="HTML")