Skip to content

base — DTOs and BaseBotOrchestrator

Base immutable objects and abstract feature orchestrator.

View DTOs

All DTOs are frozen (frozen=True). To modify, use .model_copy(update={...}).

ViewResultDTO

Bases: BaseModel

DTO for representing a single message (text + keyboard).

Attributes:

Name Type Description
text str

HTML-text of the message.
kb InlineKeyboardMarkup | None

Inline keyboard. None — without a keyboard.
Example 
view = ViewResultDTO(text="Hello!", kb=my_keyboard)
Source code in src/codex_bot/base/view_dto.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class ViewResultDTO(BaseModel):
    """DTO for representing a single message (text + keyboard).

    Attributes:
        text: HTML-text of the message.
        kb: Inline keyboard. None — without a keyboard.

    Example:
        ```python
        view = ViewResultDTO(text="Hello!", kb=my_keyboard)
        ```
    """

    text: str
    kb: InlineKeyboardMarkup | None = None

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

UnifiedViewDTO

Bases: BaseModel

Unified immutable response DTO from the Orchestrator.

Contains optional Menu and Content blocks, as well as metadata for routing and UI management (deletion, history clearing).

Attributes:

Name Type Description
content ViewResultDTO | None

Main content block (text + buttons).
menu ViewResultDTO | None

Navigation menu block. None — menu is not updated.
clean_history bool

If True — ViewSender will delete previous UI messages.
alert_text str | None

Text for a popup alert (for CallbackQuery).
trigger_message_id int | None

ID of the trigger message (e.g., /start) for deletion.
chat_id int | str | None

Target chat ID. Filled by the Director.
session_key int | str | None

Session key (user_id or channel session). Filled by the Director.
mode Literal['channel', 'topic', 'user'] | None

Sending mode — strictly "channel", "topic", or "user".
message_thread_id int | None

Topic ID in a supergroup.
Example 
view = UnifiedViewDTO(
    content=ViewResultDTO(text="Content"),
    menu=ViewResultDTO(text="Menu"),
)
# To change chat_id after creation — use model_copy:
view = view.model_copy(update={"chat_id": 123456})
Source code in src/codex_bot/base/view_dto.py 
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
class UnifiedViewDTO(BaseModel):
    """Unified immutable response DTO from the Orchestrator.

    Contains optional Menu and Content blocks, as well as metadata
    for routing and UI management (deletion, history clearing).

    Attributes:
        content: Main content block (text + buttons).
        menu: Navigation menu block. None — menu is not updated.
        clean_history: If True — ViewSender will delete previous UI messages.
        alert_text: Text for a popup alert (for CallbackQuery).
        trigger_message_id: ID of the trigger message (e.g., /start) for deletion.
        chat_id: Target chat ID. Filled by the Director.
        session_key: Session key (user_id or channel session). Filled by the Director.
        mode: Sending mode — strictly ``"channel"``, ``"topic"``, or ``"user"``.
        message_thread_id: Topic ID in a supergroup.

    Example:
        ```python
        view = UnifiedViewDTO(
            content=ViewResultDTO(text="Content"),
            menu=ViewResultDTO(text="Menu"),
        )
        # To change chat_id after creation — use model_copy:
        view = view.model_copy(update={"chat_id": 123456})
        ```
    """

    content: ViewResultDTO | None = None
    menu: ViewResultDTO | None = None
    clean_history: bool = False
    alert_text: str | None = None
    trigger_message_id: int | None = None

    # --- Routing & Session (filled by Director) ---
    chat_id: int | str | None = None
    session_key: int | str | None = None
    mode: Literal["channel", "topic", "user"] | None = None
    message_thread_id: int | None = None

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

MessageCoordsDTO

Bases: BaseModel

Telegram message coordinates (chat_id + message_id).

Attributes:

Name Type Description
chat_id int

Chat ID.
message_id int

Message ID in the chat.
Example 
coords = MessageCoordsDTO(chat_id=123456, message_id=42)
Source code in src/codex_bot/base/view_dto.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class MessageCoordsDTO(BaseModel):
    """Telegram message coordinates (chat_id + message_id).

    Attributes:
        chat_id: Chat ID.
        message_id: Message ID in the chat.

    Example:
        ```python
        coords = MessageCoordsDTO(chat_id=123456, message_id=42)
        ```
    """

    chat_id: int
    message_id: int

    model_config = ConfigDict(frozen=True)

Context DTO

BaseBotContext

Bases: BaseModel

Base immutable context for a Telegram event.

Extracted from Message or CallbackQuery via ContextHelper. Contains only identifiers — no business logic.

Attributes:

Name Type Description
user_id int

Telegram ID of the user. Used as a session key.
chat_id int

ID of the chat (private, group, channel).
message_id int | None

ID of the message that triggered the event.
message_thread_id int | None

ID of the topic in a supergroup (if applicable).
session_key int

Key for state storage (defaults to user_id).
Example 
ctx = BaseBotContext(user_id=123, chat_id=123, message_id=42)
print(ctx.session_key)  # 123
Source code in src/codex_bot/base/context_dto.py 
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
class BaseBotContext(BaseModel):
    """
    Base immutable context for a Telegram event.

    Extracted from Message or CallbackQuery via ContextHelper.
    Contains only identifiers — no business logic.

    Attributes:
        user_id: Telegram ID of the user. Used as a session key.
        chat_id: ID of the chat (private, group, channel).
        message_id: ID of the message that triggered the event.
        message_thread_id: ID of the topic in a supergroup (if applicable).
        session_key: Key for state storage (defaults to user_id).

    Example:
        ```python
        ctx = BaseBotContext(user_id=123, chat_id=123, message_id=42)
        print(ctx.session_key)  # 123
        ```
    """

    user_id: int
    chat_id: int
    message_id: int | None = None
    message_thread_id: int | None = None

    model_config = ConfigDict(frozen=True)

    @property
    def session_key(self) -> int:
        """Key for storing state in Redis (defaults to user_id)."""
        return self.user_id

Attributes

session_key property

Key for storing state in Redis (defaults to user_id).

BaseBotOrchestrator

BaseBotOrchestrator

Bases: ABC, Generic[PayloadT]

Abstract STATELESS feature orchestrator.

Defines the contract for all orchestrators in the system. The Director uses this interface for cross-feature transitions.

The class is a singleton — one instance handles requests from all users concurrently. No mutable user state in self. All context (user_id, chat_id, FSM) is passed through the director.

Subclasses must implement
  • render_content(payload, director) — main rendering logic.
Subclasses may override
  • handle_entry(director, payload) — entry point into the feature.

Parameters:

Name Type Description Default
expected_state str | None

FSM state string set when entering the feature. None — state does not change.

 None
Example 
class BookingOrchestrator(BaseBotOrchestrator[BookingPayload]):
    def __init__(self):
        super().__init__(expected_state="BookingStates:main")

    async def render_content(
        self, payload: BookingPayload, director: Director
    ) -> ViewResultDTO:
        slots = await self.api.get_slots(director.user_id)
        return ViewResultDTO(text=format_slots(slots), kb=build_kb(slots))
Source code in src/codex_bot/base/base_orchestrator.py 
 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
class BaseBotOrchestrator(ABC, Generic[PayloadT]):  # noqa: UP046
    """Abstract STATELESS feature orchestrator.

    Defines the contract for all orchestrators in the system.
    The Director uses this interface for cross-feature transitions.

    The class is a singleton — one instance handles requests from all users
    concurrently. No mutable user state in ``self``.
    All context (user_id, chat_id, FSM) is passed through the ``director``.

    Subclasses must implement:
        - ``render_content(payload, director)`` — main rendering logic.

    Subclasses may override:
        - ``handle_entry(director, payload)`` — entry point into the feature.

    Args:
        expected_state: FSM state string set when entering the feature.
                        None — state does not change.

    Example:
        ```python
        class BookingOrchestrator(BaseBotOrchestrator[BookingPayload]):
            def __init__(self):
                super().__init__(expected_state="BookingStates:main")

            async def render_content(
                self, payload: BookingPayload, director: Director
            ) -> ViewResultDTO:
                slots = await self.api.get_slots(director.user_id)
                return ViewResultDTO(text=format_slots(slots), kb=build_kb(slots))
        ```
    """

    def __init__(self, expected_state: str | None = None) -> None:
        self.expected_state = expected_state

    @abstractmethod
    async def render_content(
        self,
        payload: PayloadT,
        director: Director,
    ) -> ViewResultDTO:
        """Main logic for rendering feature content.

        Must be implemented in each specific orchestrator.

        Args:
            payload: Data for rendering (DTO from backend, dict, etc.).
            director: Context of the current request (user_id, chat_id, state).

        Returns:
            ViewResultDTO with text and keyboard.
        """
        ...

    async def handle_entry(
        self,
        director: Director,
        payload: PayloadT | None = None,
    ) -> UnifiedViewDTO:
        """Entry point into the feature. Called by the Director during set_scene().

        The default implementation simply calls render(payload, director).
        Override for complex feature initialization logic.

        Args:
            director: Context of the current request.
            payload: Initial data for rendering.

        Returns:
            UnifiedViewDTO for sending to the user.
        """
        return await self.render(payload, director)

    async def render(
        self,
        payload: PayloadT | None,
        director: Director,
    ) -> UnifiedViewDTO:
        """Assembles UnifiedViewDTO from render_content().

        Enriches the result with data from the director (chat_id, session_key).

        Args:
            payload: Data for rendering.
            director: Context of the current request.

        Returns:
            UnifiedViewDTO ready to be sent via ViewSender.
        """
        content_view = await self.render_content(payload, director)  # type: ignore[arg-type]
        return UnifiedViewDTO(content=content_view, menu=None).model_copy(
            update={
                "chat_id": director.chat_id,
                "session_key": director.user_id,
            }
        )

Functions

handle_entry(director, payload=None) async

Entry point into the feature. Called by the Director during set_scene().

The default implementation simply calls render(payload, director). Override for complex feature initialization logic.

Parameters:

Name Type Description Default
director Director

Context of the current request.

 required
payload PayloadT | None

Initial data for rendering.

 None

Returns:

Type Description
UnifiedViewDTO

UnifiedViewDTO for sending to the user.
Source code in src/codex_bot/base/base_orchestrator.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def handle_entry(
    self,
    director: Director,
    payload: PayloadT | None = None,
) -> UnifiedViewDTO:
    """Entry point into the feature. Called by the Director during set_scene().

    The default implementation simply calls render(payload, director).
    Override for complex feature initialization logic.

    Args:
        director: Context of the current request.
        payload: Initial data for rendering.

    Returns:
        UnifiedViewDTO for sending to the user.
    """
    return await self.render(payload, director)

render(payload, director) async

Assembles UnifiedViewDTO from render_content().

Enriches the result with data from the director (chat_id, session_key).

Parameters:

Name Type Description Default
payload PayloadT | None

Data for rendering.

 required
director Director

Context of the current request.

 required

Returns:

Type Description
UnifiedViewDTO

UnifiedViewDTO ready to be sent via ViewSender.
Source code in src/codex_bot/base/base_orchestrator.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def render(
    self,
    payload: PayloadT | None,
    director: Director,
) -> UnifiedViewDTO:
    """Assembles UnifiedViewDTO from render_content().

    Enriches the result with data from the director (chat_id, session_key).

    Args:
        payload: Data for rendering.
        director: Context of the current request.

    Returns:
        UnifiedViewDTO ready to be sent via ViewSender.
    """
    content_view = await self.render_content(payload, director)  # type: ignore[arg-type]
    return UnifiedViewDTO(content=content_view, menu=None).model_copy(
        update={
            "chat_id": director.chat_id,
            "session_key": director.user_id,
        }
    )

render_content(payload, director) abstractmethod async

Main logic for rendering feature content.

Must be implemented in each specific orchestrator.

Parameters:

Name Type Description Default
payload PayloadT

Data for rendering (DTO from backend, dict, etc.).

 required
director Director

Context of the current request (user_id, chat_id, state).

 required

Returns:

Type Description
ViewResultDTO

ViewResultDTO with text and keyboard.
Source code in src/codex_bot/base/base_orchestrator.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@abstractmethod
async def render_content(
    self,
    payload: PayloadT,
    director: Director,
) -> ViewResultDTO:
    """Main logic for rendering feature content.

    Must be implemented in each specific orchestrator.

    Args:
        payload: Data for rendering (DTO from backend, dict, etc.).
        director: Context of the current request (user_id, chat_id, state).

    Returns:
        ViewResultDTO with text and keyboard.
    """
    ...