Skip to content

director — Cross-feature transition coordinator

Director

Director

Coordinator of transitions between features (scenes).

Instantiated in the handler for each incoming request. Stores the request context (user_id, chat_id, state) and passes itself to orchestrators as an argument — no mutable state in the orchestrator.

Parameters:

Name Type Description Default
container ContainerProtocol

Project's DI container with a features attribute.

 required
state FSMContext | None

FSM context of the current user.

 None
user_id int | None

Telegram ID of the user.

 None
chat_id int | None

Target chat ID.

 None
trigger_id int | None

ID of the trigger message (e.g., /start) for subsequent deletion.

 None
Example 
director = Director(
    container=container,
    state=state,
    user_id=callback.from_user.id,
    chat_id=callback.message.chat.id,
)
view = await director.set_scene(feature="booking", payload=None)
await sender.send(view)
Source code in src/codex_bot/director/director.py 
 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
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class Director:
    """Coordinator of transitions between features (scenes).

    Instantiated in the handler for each incoming request.
    Stores the request context (user_id, chat_id, state) and passes itself
    to orchestrators as an argument — no mutable state in the orchestrator.

    Args:
        container: Project's DI container with a ``features`` attribute.
        state: FSM context of the current user.
        user_id: Telegram ID of the user.
        chat_id: Target chat ID.
        trigger_id: ID of the trigger message (e.g., /start) for subsequent deletion.

    Example:
        ```python
        director = Director(
            container=container,
            state=state,
            user_id=callback.from_user.id,
            chat_id=callback.message.chat.id,
        )
        view = await director.set_scene(feature="booking", payload=None)
        await sender.send(view)
        ```
    """

    def __init__(
        self,
        container: ContainerProtocol,
        state: FSMContext | None = None,
        user_id: int | None = None,
        chat_id: int | None = None,
        trigger_id: int | None = None,
    ) -> None:
        self.container = container
        self.state = state
        self.user_id = user_id
        self.chat_id = chat_id
        self.trigger_id = trigger_id

    async def set_scene(self, feature: str, payload: Any = None) -> Any:
        """Cross-feature transition: changes FSM state and calls the feature orchestrator.

        Algorithm:
        1. Retrieves the orchestrator by key from ``container.features``.
        2. Sets the FSM state if the orchestrator has declared it.
        3. Passes itself to ``handle_entry(director=self, payload=payload)``.
        4. Enriches the result with ``chat_id`` and ``session_key`` as a fallback.

        Args:
            feature: Orchestrator key in ``container.features`` (e.g., ``"booking"``).
            payload: Data to pass to ``handle_entry()``.

        Returns:
            UnifiedViewDTO or any orchestrator result. None if the feature is not found.
        """
        orchestrator = self.container.features.get(feature)

        if orchestrator is None:
            log.error(f"Director | unknown_feature='{feature}' user_id={self.user_id}")
            return None

        # 1. FSM state change
        if self.state and hasattr(orchestrator, "expected_state") and orchestrator.expected_state:
            await self.state.set_state(orchestrator.expected_state)

        # 2. Call handle_entry (pass self as context) or render
        if isinstance(orchestrator, OrchestratorProtocol):
            view = await orchestrator.handle_entry(director=self, payload=payload)
        elif hasattr(orchestrator, "render"):
            view = await orchestrator.render(payload, self)
        else:
            log.warning(f"Director | orchestrator='{feature}' has no handle_entry or render")
            return None

        # 3. Fallback enrichment of UnifiedViewDTO with session data
        if isinstance(view, UnifiedViewDTO):
            view = view.model_copy(
                update={
                    "chat_id": view.chat_id or self.chat_id,
                    "session_key": view.session_key or self.user_id,
                }
            )

        return view

Functions

set_scene(feature, payload=None) async

Cross-feature transition: changes FSM state and calls the feature orchestrator.

Algorithm: 1. Retrieves the orchestrator by key from container.features. 2. Sets the FSM state if the orchestrator has declared it. 3. Passes itself to handle_entry(director=self, payload=payload). 4. Enriches the result with chat_id and session_key as a fallback.

Parameters:

Name Type Description Default
feature str

Orchestrator key in container.features (e.g., "booking").

 required
payload Any

Data to pass to handle_entry().

 None

Returns:

Type Description
Any

UnifiedViewDTO or any orchestrator result. None if the feature is not found.
Source code in src/codex_bot/director/director.py 
 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
async def set_scene(self, feature: str, payload: Any = None) -> Any:
    """Cross-feature transition: changes FSM state and calls the feature orchestrator.

    Algorithm:
    1. Retrieves the orchestrator by key from ``container.features``.
    2. Sets the FSM state if the orchestrator has declared it.
    3. Passes itself to ``handle_entry(director=self, payload=payload)``.
    4. Enriches the result with ``chat_id`` and ``session_key`` as a fallback.

    Args:
        feature: Orchestrator key in ``container.features`` (e.g., ``"booking"``).
        payload: Data to pass to ``handle_entry()``.

    Returns:
        UnifiedViewDTO or any orchestrator result. None if the feature is not found.
    """
    orchestrator = self.container.features.get(feature)

    if orchestrator is None:
        log.error(f"Director | unknown_feature='{feature}' user_id={self.user_id}")
        return None

    # 1. FSM state change
    if self.state and hasattr(orchestrator, "expected_state") and orchestrator.expected_state:
        await self.state.set_state(orchestrator.expected_state)

    # 2. Call handle_entry (pass self as context) or render
    if isinstance(orchestrator, OrchestratorProtocol):
        view = await orchestrator.handle_entry(director=self, payload=payload)
    elif hasattr(orchestrator, "render"):
        view = await orchestrator.render(payload, self)
    else:
        log.warning(f"Director | orchestrator='{feature}' has no handle_entry or render")
        return None

    # 3. Fallback enrichment of UnifiedViewDTO with session data
    if isinstance(view, UnifiedViewDTO):
        view = view.model_copy(
            update={
                "chat_id": view.chat_id or self.chat_id,
                "session_key": view.session_key or self.user_id,
            }
        )

    return view

Protocols

OrchestratorProtocol

Bases: Protocol

Minimum contract for a stateless feature orchestrator.

The Director works through this protocol without knowing about specific classes. BaseBotOrchestrator implements it automatically.

The orchestrator must be stateless — it does not store user state. Context is passed via the director argument on each call.

Source code in src/codex_bot/director/protocols.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@runtime_checkable
class OrchestratorProtocol(Protocol):
    """Minimum contract for a stateless feature orchestrator.

    The Director works through this protocol without knowing about specific classes.
    BaseBotOrchestrator implements it automatically.

    The orchestrator must be stateless — it does not store user state.
    Context is passed via the ``director`` argument on each call.
    """

    async def render(self, payload: Any, director: Any) -> Any:
        """Renders content for the passed payload."""
        ...

    async def handle_entry(
        self,
        director: Any,
        payload: Any = None,
    ) -> Any:
        """Entry point into the feature."""
        ...

Functions

handle_entry(director, payload=None) async

Entry point into the feature.

Source code in src/codex_bot/director/protocols.py 
53
54
55
56
57
58
59
async def handle_entry(
    self,
    director: Any,
    payload: Any = None,
) -> Any:
    """Entry point into the feature."""
    ...

render(payload, director) async

Renders content for the passed payload.

Source code in src/codex_bot/director/protocols.py 
49
50
51
async def render(self, payload: Any, director: Any) -> Any:
    """Renders content for the passed payload."""
    ...

ContainerProtocol

Bases: Protocol

Minimum contract for the project's DI container.

The Director only requires the features attribute — a dictionary of orchestrators. The specific BotContainer of the project must provide this attribute.

Attributes:

Name Type Description
features dict[str, OrchestratorProtocol]

Dictionary of {feature_key: orchestrator}.
Example 
class BotContainer:
    def __init__(self):
        self.features: dict[str, Any] = {}
Source code in src/codex_bot/director/protocols.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@runtime_checkable
class ContainerProtocol(Protocol):
    """Minimum contract for the project's DI container.

    The Director only requires the ``features`` attribute — a dictionary of orchestrators.
    The specific BotContainer of the project must provide this attribute.

    Attributes:
        features: Dictionary of ``{feature_key: orchestrator}``.

    Example:
        ```python
        class BotContainer:
            def __init__(self):
                self.features: dict[str, Any] = {}
        ```
    """

    features: dict[str, OrchestratorProtocol]

SceneConfig

Bases: NamedTuple

Scene configuration: FSM state + entry-point service key.

Used in the project's SCENE_ROUTES to describe cross-feature transitions.

Attributes:

Name Type Description
fsm_state State

Aiogram State set when entering the scene.
entry_service str

Orchestrator key in the container registry (e.g., "booking").
Example 
SCENE_ROUTES = {
    "booking": SceneConfig(
        fsm_state=BookingStates.main,
        entry_service="booking",
    ),
}
Source code in src/codex_bot/director/protocols.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class SceneConfig(NamedTuple):
    """Scene configuration: FSM state + entry-point service key.

    Used in the project's SCENE_ROUTES to describe cross-feature transitions.

    Attributes:
        fsm_state: Aiogram State set when entering the scene.
        entry_service: Orchestrator key in the container registry (e.g., ``"booking"``).

    Example:
        ```python
        SCENE_ROUTES = {
            "booking": SceneConfig(
                fsm_state=BookingStates.main,
                entry_service="booking",
            ),
        }
        ```
    """

    fsm_state: State
    entry_service: str