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

Data Transfer Object representing a single Telegram message.

Encapsulates the visual components of a message, including its textual content and interactive elements. Designed for use within UnifiedViewDTO.

Attributes:

Name Type Description
text str

HTML-formatted string containing the message body.
kb InlineKeyboardMarkup | None

Optional inline keyboard markup for user interaction.
Example 
view = ViewResultDTO(text="<b>Success!</b>", kb=inline_kb)
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
32
33
34
class ViewResultDTO(BaseModel):
    """Data Transfer Object representing a single Telegram message.

    Encapsulates the visual components of a message, including its textual
    content and interactive elements. Designed for use within `UnifiedViewDTO`.

    Attributes:
        text: HTML-formatted string containing the message body.
        kb: Optional inline keyboard markup for user interaction.

    Example:
        ```python
        view = ViewResultDTO(text="<b>Success!</b>", kb=inline_kb)
        ```
    """

    text: str
    kb: InlineKeyboardMarkup | None = None

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

UnifiedViewDTO

Bases: BaseModel

Unified response DTO for cross-service UI synchronization.

This DTO serves as the primary contract between orchestrators and the ViewSender. It aggregates multiple UI components (menus, content) and routing metadata required for complex interaction flows.

Attributes:

Name Type Description
content ViewResultDTO | None

Primary content block representing the main response message.
menu ViewResultDTO | None

Optional navigation menu block for persistent UI elements.
clean_history bool

If True, instructs the sender to prune previous UI messages.
alert_text str | None

Plain text for CallbackQuery toast notifications.
trigger_message_id int | None

ID of the message that initiated the request (for cleanup).
chat_id int | str | None

Target destination for the UI update. Populated by the Director.
session_key int | str | None

Identifier for user or channel session isolation.
mode Literal['channel', 'topic', 'user'] | None

Telegram delivery mode (channel/topic/user).
message_thread_id int | None

Destination thread ID for supergroup topics.
Example 
view = UnifiedViewDTO(
    content=ViewResultDTO(text="Operation completed."),
    clean_history=True
)
Source code in src/codex_bot/base/view_dto.py 
 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
class UnifiedViewDTO(BaseModel):
    """Unified response DTO for cross-service UI synchronization.

    This DTO serves as the primary contract between orchestrators and the
    `ViewSender`. It aggregates multiple UI components (menus, content)
    and routing metadata required for complex interaction flows.

    Attributes:
        content: Primary content block representing the main response message.
        menu: Optional navigation menu block for persistent UI elements.
        clean_history: If True, instructs the sender to prune previous UI messages.
        alert_text: Plain text for CallbackQuery toast notifications.
        trigger_message_id: ID of the message that initiated the request (for cleanup).
        chat_id: Target destination for the UI update. Populated by the `Director`.
        session_key: Identifier for user or channel session isolation.
        mode: Telegram delivery mode (channel/topic/user).
        message_thread_id: Destination thread ID for supergroup topics.

    Example:
        ```python
        view = UnifiedViewDTO(
            content=ViewResultDTO(text="Operation completed."),
            clean_history=True
        )
        ```
    """

    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

    async def send(self, message: Any, i18n: Any = None) -> Any:
        """
        Backward compatibility bridge for generated handlers.
        Sends the content part of the DTO to the user.
        """
        if not self.content:
            return None

        return await message.answer(text=self.content.text, reply_markup=self.content.kb, parse_mode="HTML")

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

Functions

send(message, i18n=None) async

Backward compatibility bridge for generated handlers. Sends the content part of the DTO to the user.

Source code in src/codex_bot/base/view_dto.py 
 95
 96
 97
 98
 99
100
101
102
103
async def send(self, message: Any, i18n: Any = None) -> Any:
    """
    Backward compatibility bridge for generated handlers.
    Sends the content part of the DTO to the user.
    """
    if not self.content:
        return None

    return await message.answer(text=self.content.text, reply_markup=self.content.kb, parse_mode="HTML")

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 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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 base class for feature-specific orchestrators.

Implements the OrchestratorProtocol and provides a template for rendering UI responses. Orchestrators are designed as stateless services that operate on a request-scoped Director context.

Note

Subclasses must be thread-safe for use in asyncio loops as they are typically registered as singletons in the DI container.

Parameters:

Name Type Description Default
expected_state str | None

The FSM state string to be set by the Director when this feature is entered. If None, the state remains unchanged.

 None
Example 
class ProfileOrchestrator(BaseBotOrchestrator[ProfilePayload]):
    async def render_content(self, director, payload):
        return ViewResultDTO(text=f"User: {payload.name}")
Source code in src/codex_bot/base/base_orchestrator.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
class BaseBotOrchestrator(ABC, Generic[PayloadT]):  # noqa: UP046
    """Abstract base class for feature-specific orchestrators.

    Implements the `OrchestratorProtocol` and provides a template for
    rendering UI responses. Orchestrators are designed as stateless services
    that operate on a request-scoped `Director` context.

    Note:
        Subclasses must be thread-safe for use in `asyncio` loops as they are
        typically registered as singletons in the DI container.

    Args:
        expected_state: The FSM state string to be set by the `Director`
            when this feature is entered. If None, the state remains unchanged.

    Example:
        ```python
        class ProfileOrchestrator(BaseBotOrchestrator[ProfilePayload]):
            async def render_content(self, director, payload):
                return ViewResultDTO(text=f"User: {payload.name}")
        ```
    """

    access_scopes: set[str] = set()
    """Set of required scopes for this feature. Used by TransitionGuards."""

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

    @abstractmethod
    async def render_content(
        self,
        director: Director | None = None,
        payload: PayloadT | None = None,
    ) -> ViewResultDTO | UnifiedViewDTO:
        """Analyze business data and generate the primary UI components.

        This is the primary extension point for feature logic. It must process
         the payload and return the visual representation of the feature.

        Args:
            director: Request-scoped coordinator providing access to DI and session.
            payload: Domain-specific data required for rendering.

        Returns:
            A `ViewResultDTO` for standard responses or `UnifiedViewDTO` for
            complex responses or redirects.
        """
        ...

    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(director, payload).
        Override for complex feature initialization logic.

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

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

    async def render(
        self,
        director: Director | None = None,
        payload: PayloadT | None = None,
    ) -> UnifiedViewDTO:
        """Assemble a `UnifiedViewDTO` by executing the rendering logic.

        Wraps the result of `render_content` into a standardized delivery
        format. Handles automatic enrichment for standard response types.

        Args:
            director: Current request context.
            payload: Data to be rendered.

        Returns:
            A fully structured `UnifiedViewDTO` ready for transmission.
        """
        # Internal library calls now strictly use keyword arguments for safety
        view = await self.render_content(director=director, payload=payload)

        # If render_content already returned a full UnifiedViewDTO (Redirect)
        res = view if isinstance(view, UnifiedViewDTO) else UnifiedViewDTO(content=view, menu=None)

        # Session enrichment (chat_id, session_key) from the Director context
        if director:
            res = res.model_copy(
                update={
                    "chat_id": res.chat_id or director.context_id,
                    "session_key": res.session_key or director.session_key,
                    "trigger_message_id": res.trigger_message_id or director.trigger_id,
                }
            )

        return res

Attributes

access_scopes = set() class-attribute instance-attribute

Set of required scopes for this feature. Used by TransitionGuards.

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(director, payload). Override for complex feature initialization logic.

Parameters:

Name Type Description Default
director Director

Context of the 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 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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(director, payload).
    Override for complex feature initialization logic.

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

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

render(director=None, payload=None) async

Assemble a UnifiedViewDTO by executing the rendering logic.

Wraps the result of render_content into a standardized delivery format. Handles automatic enrichment for standard response types.

Parameters:

Name Type Description Default
director Director | None

Current request context.

 None
payload PayloadT | None

Data to be rendered.

 None

Returns:

Type Description
UnifiedViewDTO

A fully structured UnifiedViewDTO ready for transmission.
Source code in src/codex_bot/base/base_orchestrator.py 
 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
async def render(
    self,
    director: Director | None = None,
    payload: PayloadT | None = None,
) -> UnifiedViewDTO:
    """Assemble a `UnifiedViewDTO` by executing the rendering logic.

    Wraps the result of `render_content` into a standardized delivery
    format. Handles automatic enrichment for standard response types.

    Args:
        director: Current request context.
        payload: Data to be rendered.

    Returns:
        A fully structured `UnifiedViewDTO` ready for transmission.
    """
    # Internal library calls now strictly use keyword arguments for safety
    view = await self.render_content(director=director, payload=payload)

    # If render_content already returned a full UnifiedViewDTO (Redirect)
    res = view if isinstance(view, UnifiedViewDTO) else UnifiedViewDTO(content=view, menu=None)

    # Session enrichment (chat_id, session_key) from the Director context
    if director:
        res = res.model_copy(
            update={
                "chat_id": res.chat_id or director.context_id,
                "session_key": res.session_key or director.session_key,
                "trigger_message_id": res.trigger_message_id or director.trigger_id,
            }
        )

    return res

render_content(director=None, payload=None) abstractmethod async

Analyze business data and generate the primary UI components.

This is the primary extension point for feature logic. It must process the payload and return the visual representation of the feature.

Parameters:

Name Type Description Default
director Director | None

Request-scoped coordinator providing access to DI and session.

 None
payload PayloadT | None

Domain-specific data required for rendering.

 None

Returns:

Type Description
ViewResultDTO | UnifiedViewDTO

A ViewResultDTO for standard responses or UnifiedViewDTO for
ViewResultDTO | UnifiedViewDTO

complex responses or redirects.
Source code in src/codex_bot/base/base_orchestrator.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@abstractmethod
async def render_content(
    self,
    director: Director | None = None,
    payload: PayloadT | None = None,
) -> ViewResultDTO | UnifiedViewDTO:
    """Analyze business data and generate the primary UI components.

    This is the primary extension point for feature logic. It must process
     the payload and return the visual representation of the feature.

    Args:
        director: Request-scoped coordinator providing access to DI and session.
        payload: Domain-specific data required for rendering.

    Returns:
        A `ViewResultDTO` for standard responses or `UnifiedViewDTO` for
        complex responses or redirects.
    """
    ...