Skip to content

engine.factory — Bot + Dispatcher creation

BotBuilder

BotBuilder

Builder for Bot + Dispatcher with automatic core middleware management.

Ensures that essential framework middlewares are registered in the correct sequence. Automatically injects the created Bot instance into the provided Container.

Parameters:

Name Type Description Default
bot_token str

Telegram bot token obtained from @BotFather.

 required
parse_mode str

Message parsing mode (default is "HTML").

 'HTML'
fsm_storage BaseStorage | None

FSM storage. None → MemoryStorage.

 None
dispatcher_kwargs dict[str, Any] | None

Additional kwargs for Dispatcher initialization.

 None
Example 
builder = BotBuilder(bot_token="...")
builder.setup_core(container=my_container)
builder.add_project_middleware(MyAnalyticsMiddleware())

bot, dp = builder.build()
Source code in src/codex_bot/engine/factory/bot_builder.py 
 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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class BotBuilder:
    """Builder for Bot + Dispatcher with automatic core middleware management.

    Ensures that essential framework middlewares are registered in the correct sequence.
    Automatically injects the created ``Bot`` instance into the provided ``Container``.

    Args:
        bot_token: Telegram bot token obtained from @BotFather.
        parse_mode: Message parsing mode (default is ``"HTML"``).
        fsm_storage: FSM storage. ``None`` → ``MemoryStorage``.
        dispatcher_kwargs: Additional kwargs for ``Dispatcher`` initialization.

    Example:
        ```python
        builder = BotBuilder(bot_token="...")
        builder.setup_core(container=my_container)
        builder.add_project_middleware(MyAnalyticsMiddleware())

        bot, dp = builder.build()
        ```
    """

    def __init__(
        self,
        bot_token: str,
        parse_mode: str = "HTML",
        fsm_storage: BaseStorage | None = None,
        dispatcher_kwargs: dict[str, Any] | None = None,
    ) -> None:
        """Initializes the builder with basic settings."""
        if not bot_token:
            raise ValueError("bot_token must not be empty")

        self._bot_token = bot_token
        self._parse_mode = parse_mode
        self._fsm_storage: BaseStorage = fsm_storage or MemoryStorage()
        self._dispatcher_kwargs: dict[str, Any] = dispatcher_kwargs or {}

        self._core_middlewares: list[BaseMiddleware] = []
        self._project_middlewares: list[BaseMiddleware] = []
        self._middlewares = self._project_middlewares  # Alias for backward compatibility in tests
        self._container: ContainerProtocol | None = None

    def setup_core(self, container: ContainerProtocol) -> BotBuilder:
        """Registers mandatory framework middlewares and links the container.

        Registration order:
        1. **Container Injection**: Makes the DI container available in context.
        2. **Director Initialization**: Creates the request coordinator.
        3. **User Validation (RBAC)**: Handles user identification and admin checks.

        Args:
            container: The project's DI container implementing ContainerProtocol.

        Returns:
            The builder instance for method chaining.
        """
        self._container = container

        # Internal core middleware setup
        self._core_middlewares = [
            ContainerMiddleware(container),
            DirectorMiddleware(),
            UserValidationMiddleware(container),
        ]
        return self

    def add_project_middleware(self, middleware: BaseMiddleware) -> BotBuilder:
        """Adds a project-specific middleware to the execution chain.

        Project middlewares are automatically registered AFTER the core framework ones.

        Args:
            middleware: Instance of an aiogram BaseMiddleware.

        Returns:
            The builder instance for method chaining.
        """
        self._project_middlewares.append(middleware)
        return self

    def add_middleware(self, middleware: BaseMiddleware) -> BotBuilder:
        """Alias for add_project_middleware. Maintained for backward compatibility."""
        return self.add_project_middleware(middleware)

    def build(self) -> tuple[Bot, Dispatcher]:
        """Assembles and configures Bot and Dispatcher.

        1. Creates the Bot instance.
        2. Injects the Bot into the Container (if linked via setup_core).
        3. Initializes the Dispatcher and registers all middlewares.

        Returns:
            A tuple of (Bot, Dispatcher).
        """
        bot = Bot(
            token=self._bot_token,
            default=DefaultBotProperties(parse_mode=self._parse_mode),
        )

        # Automatic Integration: Linking Bot and Container
        if self._container:
            self._container.set_bot(bot)
            log.debug("BotBuilder | Automatically injected Bot instance into Container")

        dp = Dispatcher(storage=self._fsm_storage, **self._dispatcher_kwargs)

        # Register Core Middlewares first (Outer level)
        for mw in self._core_middlewares:
            dp.update.outer_middleware(mw)
            log.debug(f"BotBuilder | Core Middleware: {mw.__class__.__name__}")

        # Register Project Middlewares
        for mw in self._project_middlewares:
            dp.update.outer_middleware(mw)
            log.debug(f"BotBuilder | Project Middleware: {mw.__class__.__name__}")

        log.info(
            f"BotBuilder | Built with {len(self._core_middlewares)} core "
            f"and {len(self._project_middlewares)} project middlewares"
        )
        return bot, dp

Functions

__init__(bot_token, parse_mode='HTML', fsm_storage=None, dispatcher_kwargs=None)

Initializes the builder with basic settings.

Source code in src/codex_bot/engine/factory/bot_builder.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    bot_token: str,
    parse_mode: str = "HTML",
    fsm_storage: BaseStorage | None = None,
    dispatcher_kwargs: dict[str, Any] | None = None,
) -> None:
    """Initializes the builder with basic settings."""
    if not bot_token:
        raise ValueError("bot_token must not be empty")

    self._bot_token = bot_token
    self._parse_mode = parse_mode
    self._fsm_storage: BaseStorage = fsm_storage or MemoryStorage()
    self._dispatcher_kwargs: dict[str, Any] = dispatcher_kwargs or {}

    self._core_middlewares: list[BaseMiddleware] = []
    self._project_middlewares: list[BaseMiddleware] = []
    self._middlewares = self._project_middlewares  # Alias for backward compatibility in tests
    self._container: ContainerProtocol | None = None

add_middleware(middleware)

Alias for add_project_middleware. Maintained for backward compatibility.

Source code in src/codex_bot/engine/factory/bot_builder.py 
108
109
110
def add_middleware(self, middleware: BaseMiddleware) -> BotBuilder:
    """Alias for add_project_middleware. Maintained for backward compatibility."""
    return self.add_project_middleware(middleware)

add_project_middleware(middleware)

Adds a project-specific middleware to the execution chain.

Project middlewares are automatically registered AFTER the core framework ones.

Parameters:

Name Type Description Default
middleware BaseMiddleware

Instance of an aiogram BaseMiddleware.

 required

Returns:

Type Description
BotBuilder

The builder instance for method chaining.
Source code in src/codex_bot/engine/factory/bot_builder.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def add_project_middleware(self, middleware: BaseMiddleware) -> BotBuilder:
    """Adds a project-specific middleware to the execution chain.

    Project middlewares are automatically registered AFTER the core framework ones.

    Args:
        middleware: Instance of an aiogram BaseMiddleware.

    Returns:
        The builder instance for method chaining.
    """
    self._project_middlewares.append(middleware)
    return self

build()

Assembles and configures Bot and Dispatcher.

  1. Creates the Bot instance.
  2. Injects the Bot into the Container (if linked via setup_core).
  3. Initializes the Dispatcher and registers all middlewares.

Returns:

Type Description
tuple[Bot, Dispatcher]

A tuple of (Bot, Dispatcher).
Source code in src/codex_bot/engine/factory/bot_builder.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def build(self) -> tuple[Bot, Dispatcher]:
    """Assembles and configures Bot and Dispatcher.

    1. Creates the Bot instance.
    2. Injects the Bot into the Container (if linked via setup_core).
    3. Initializes the Dispatcher and registers all middlewares.

    Returns:
        A tuple of (Bot, Dispatcher).
    """
    bot = Bot(
        token=self._bot_token,
        default=DefaultBotProperties(parse_mode=self._parse_mode),
    )

    # Automatic Integration: Linking Bot and Container
    if self._container:
        self._container.set_bot(bot)
        log.debug("BotBuilder | Automatically injected Bot instance into Container")

    dp = Dispatcher(storage=self._fsm_storage, **self._dispatcher_kwargs)

    # Register Core Middlewares first (Outer level)
    for mw in self._core_middlewares:
        dp.update.outer_middleware(mw)
        log.debug(f"BotBuilder | Core Middleware: {mw.__class__.__name__}")

    # Register Project Middlewares
    for mw in self._project_middlewares:
        dp.update.outer_middleware(mw)
        log.debug(f"BotBuilder | Project Middleware: {mw.__class__.__name__}")

    log.info(
        f"BotBuilder | Built with {len(self._core_middlewares)} core "
        f"and {len(self._project_middlewares)} project middlewares"
    )
    return bot, dp

setup_core(container)

Registers mandatory framework middlewares and links the container.

Registration order: 1. Container Injection: Makes the DI container available in context. 2. Director Initialization: Creates the request coordinator. 3. User Validation (RBAC): Handles user identification and admin checks.

Parameters:

Name Type Description Default
container ContainerProtocol

The project's DI container implementing ContainerProtocol.

 required

Returns:

Type Description
BotBuilder

The builder instance for method chaining.
Source code in src/codex_bot/engine/factory/bot_builder.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def setup_core(self, container: ContainerProtocol) -> BotBuilder:
    """Registers mandatory framework middlewares and links the container.

    Registration order:
    1. **Container Injection**: Makes the DI container available in context.
    2. **Director Initialization**: Creates the request coordinator.
    3. **User Validation (RBAC)**: Handles user identification and admin checks.

    Args:
        container: The project's DI container implementing ContainerProtocol.

    Returns:
        The builder instance for method chaining.
    """
    self._container = container

    # Internal core middleware setup
    self._core_middlewares = [
        ContainerMiddleware(container),
        DirectorMiddleware(),
        UserValidationMiddleware(container),
    ]
    return self