Skip to content

engine.discovery — Feature discovery

FeatureDiscoveryService

FeatureDiscoveryService

Service for discovering and registering feature configurations.

Two modes:

Mode 1: Auto-discovery (similar to Django's INSTALLED_APPS):

discovery = FeatureDiscoveryService(
    module_prefix="src.telegram_bot",
    installed_features=["features.telegram.commands", "features.telegram.booking"],
    installed_redis_features=["features.redis.notifications"],
    redis_dispatcher=bot_redis_dispatcher,
)
discovery.discover_all()
orchestrators = discovery.create_feature_orchestrators(container)

Mode 2: Explicit registration (fallback):

discovery = FeatureDiscoveryService()
discovery.register_router(commands_router)
discovery.register_orchestrator("booking", BookingOrchestrator(container))
discovery.register_garbage_states(BookingStates)

Parameters:

Name Type Description Default
module_prefix str

Module path prefix for auto-discovery. Example: "src.telegram_bot".

 ''
installed_features list[str] | None

List of Telegram feature paths (with Aiogram routers).

 None
installed_redis_features list[str] | None

List of Redis feature paths (with RedisRouter).

 None
redis_dispatcher BotRedisDispatcher | None

Dispatcher for registering Redis handlers.

 None
Source code in src/codex_bot/engine/discovery/service.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
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class FeatureDiscoveryService:
    """
    Service for discovering and registering feature configurations.

    Two modes:

    **Mode 1: Auto-discovery** (similar to Django's INSTALLED_APPS):
    ```python
    discovery = FeatureDiscoveryService(
        module_prefix="src.telegram_bot",
        installed_features=["features.telegram.commands", "features.telegram.booking"],
        installed_redis_features=["features.redis.notifications"],
        redis_dispatcher=bot_redis_dispatcher,
    )
    discovery.discover_all()
    orchestrators = discovery.create_feature_orchestrators(container)
    ```

    **Mode 2: Explicit registration** (fallback):
    ```python
    discovery = FeatureDiscoveryService()
    discovery.register_router(commands_router)
    discovery.register_orchestrator("booking", BookingOrchestrator(container))
    discovery.register_garbage_states(BookingStates)
    ```

    Args:
        module_prefix: Module path prefix for auto-discovery.
                       Example: "src.telegram_bot".
        installed_features: List of Telegram feature paths (with Aiogram routers).
        installed_redis_features: List of Redis feature paths (with RedisRouter).
        redis_dispatcher: Dispatcher for registering Redis handlers.
    """

    def __init__(
        self,
        module_prefix: str = "",
        installed_features: list[str] | None = None,
        installed_redis_features: list[str] | None = None,
        redis_dispatcher: BotRedisDispatcher | None = None,
    ) -> None:
        self._prefix = module_prefix
        self._features = installed_features or []
        self._redis_features = installed_redis_features or []
        self._redis_dispatcher = redis_dispatcher

        # Explicitly registered objects (Mode 2)
        self._explicit_routers: list[Router] = []
        self._explicit_orchestrators: dict[str, Any] = {}

    # =========================================================================
    # Mode 1: Auto-discovery
    # =========================================================================

    def discover_all(self) -> None:
        """
        Starts auto-discovery of all registered features.

        For Telegram features: loads menu config, garbage states, Aiogram routers.
        For Redis features: loads RedisRouter and connects it to the dispatcher.
        """
        for feature_path in self._features:
            self._discover_menu(feature_path)
            self._discover_garbage_states(feature_path)

        for feature_path in self._redis_features:
            self._discover_redis_handlers(feature_path)
            self._discover_garbage_states(feature_path)

    def create_feature_orchestrators(self, container: Any) -> dict[str, Any]:
        """
        Creates orchestrators for all features via their factory functions.

        Looks for the `create_orchestrator(container)` function in the `feature_setting.py`
        of each feature. The key prefix for Redis features is "redis_".

        Args:
            container: Project's DI container.

        Returns:
            Dictionary {feature_key: orchestrator_instance}.
        """
        orchestrators = dict(self._explicit_orchestrators)
        configs = [(self._features, ""), (self._redis_features, "redis_")]

        for feature_list, prefix in configs:
            for feature_path in feature_list:
                module = self._load_feature_module(feature_path)
                if not module:
                    continue
                factory = getattr(module, "create_orchestrator", None)
                if not factory:
                    continue
                base_name = feature_path.split(".")[-1]
                key = f"{prefix}{base_name}"
                orchestrators[key] = factory(container)
                log.info(f"FeatureDiscovery | orchestrator loaded key='{key}'")

        return orchestrators

    def collect_aiogram_routers(self) -> list[Router]:
        """
        Collects Aiogram Routers from all Telegram features.

        Returns:
            List of Routers to be included in the main router.
        """
        routers = list(self._explicit_routers)

        for feature_path in self._features:
            module_path = f"{self._prefix}.{feature_path}.handlers" if self._prefix else f"{feature_path}.handlers"
            try:
                module = importlib.import_module(module_path)
                router = getattr(module, "router", None)
                if router and isinstance(router, Router):
                    routers.append(router)
                    log.info(f"FeatureDiscovery | router loaded feature='{feature_path}'")
            except ImportError as e:
                if getattr(e, "name", None) == module_path:
                    log.debug(f"FeatureDiscovery | no handlers file feature='{feature_path}'")
                else:
                    log.critical(f"FeatureDiscovery | Broken import inside '{feature_path}': {e}")
                    raise

        return routers

    def get_menu_buttons(self, is_admin: bool | None = None) -> dict[str, dict[str, Any]]:
        """
        Returns menu button configurations for all Telegram features.

        Args:
            is_admin: None — all buttons, True — only admin, False — only user.

        Returns:
            Dictionary {feature_key: menu_config_dict}.
        """
        buttons: dict[str, dict[str, Any]] = {}
        for feature_path in self._features:
            btn = self._discover_menu(feature_path)
            if btn:
                if is_admin is not None and btn.get("is_admin", False) != is_admin:
                    continue
                key = btn.get("key", feature_path)
                buttons[key] = btn
        return buttons

    # =========================================================================
    # Mode 2: Explicit registration (fallback)
    # =========================================================================

    def register_router(self, router: Router) -> None:
        """
        Explicitly registers an Aiogram Router (without auto-discovery).

        Args:
            router: Instance of aiogram.Router.
        """
        self._explicit_routers.append(router)

    def register_orchestrator(self, key: str, orchestrator: Any) -> None:
        """
        Explicitly registers a feature orchestrator.

        Args:
            key: Feature key (e.g., "booking", "redis_notifications").
            orchestrator: Orchestrator instance.
        """
        self._explicit_orchestrators[key] = orchestrator

    def register_garbage_states(self, states: Any) -> None:
        """
        Explicitly registers states as garbage (Garbage Collector).

        Args:
            states: State, StatesGroup, string, or a list of them.
        """
        GarbageStateRegistry.register(states)

    # =========================================================================
    # Private auto-discovery methods
    # =========================================================================

    def _load_feature_module(self, feature_path: str) -> ModuleType | None:
        candidates = []
        if self._prefix:
            candidates.append(f"{self._prefix}.{feature_path}.feature_setting")
            candidates.append(f"{self._prefix}.{feature_path}")
        candidates.append(f"{feature_path}.feature_setting")
        candidates.append(feature_path)

        for path in candidates:
            try:
                return importlib.import_module(path)
            except ImportError:
                continue
        return None

    def _discover_menu(self, feature_path: str) -> dict[str, Any] | None:
        module = self._load_feature_module(feature_path)
        if module:
            config = getattr(module, "MENU_CONFIG", None)
            if config and isinstance(config, dict):
                return cast(dict[str, Any] | None, config)
        return None

    def _discover_garbage_states(self, feature_path: str) -> None:
        module = self._load_feature_module(feature_path)
        if not module:
            return
        garbage = getattr(module, "GARBAGE_STATES", None)
        if garbage:
            GarbageStateRegistry.register(garbage)
            return
        if getattr(module, "GARBAGE_COLLECT", False):
            states = getattr(module, "STATES", None)
            if states:
                GarbageStateRegistry.register(states)

    def _discover_redis_handlers(self, feature_path: str) -> None:
        if not self._redis_dispatcher:
            return
        module_path = f"{self._prefix}.{feature_path}.handlers" if self._prefix else f"{feature_path}.handlers"
        try:
            module = importlib.import_module(module_path)
            redis_router = getattr(module, "redis_router", None)
            if redis_router and isinstance(redis_router, RedisRouter):
                self._redis_dispatcher.include_router(redis_router)
        except ImportError as e:
            if getattr(e, "name", None) == module_path:
                log.debug(f"FeatureDiscovery | no redis handlers file feature='{module_path}'")
            else:
                log.critical(f"FeatureDiscovery | Broken import inside '{module_path}': {e}")
                raise

Functions

collect_aiogram_routers()

Collects Aiogram Routers from all Telegram features.

Returns:

Type Description
list[Router]

List of Routers to be included in the main router.
Source code in src/codex_bot/engine/discovery/service.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def collect_aiogram_routers(self) -> list[Router]:
    """
    Collects Aiogram Routers from all Telegram features.

    Returns:
        List of Routers to be included in the main router.
    """
    routers = list(self._explicit_routers)

    for feature_path in self._features:
        module_path = f"{self._prefix}.{feature_path}.handlers" if self._prefix else f"{feature_path}.handlers"
        try:
            module = importlib.import_module(module_path)
            router = getattr(module, "router", None)
            if router and isinstance(router, Router):
                routers.append(router)
                log.info(f"FeatureDiscovery | router loaded feature='{feature_path}'")
        except ImportError as e:
            if getattr(e, "name", None) == module_path:
                log.debug(f"FeatureDiscovery | no handlers file feature='{feature_path}'")
            else:
                log.critical(f"FeatureDiscovery | Broken import inside '{feature_path}': {e}")
                raise

    return routers

create_feature_orchestrators(container)

Creates orchestrators for all features via their factory functions.

Looks for the create_orchestrator(container) function in the feature_setting.py of each feature. The key prefix for Redis features is "redis_".

Parameters:

Name Type Description Default
container Any

Project's DI container.

 required

Returns:

Type Description
dict[str, Any]

Dictionary {feature_key: orchestrator_instance}.
Source code in src/codex_bot/engine/discovery/service.py 
 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
def create_feature_orchestrators(self, container: Any) -> dict[str, Any]:
    """
    Creates orchestrators for all features via their factory functions.

    Looks for the `create_orchestrator(container)` function in the `feature_setting.py`
    of each feature. The key prefix for Redis features is "redis_".

    Args:
        container: Project's DI container.

    Returns:
        Dictionary {feature_key: orchestrator_instance}.
    """
    orchestrators = dict(self._explicit_orchestrators)
    configs = [(self._features, ""), (self._redis_features, "redis_")]

    for feature_list, prefix in configs:
        for feature_path in feature_list:
            module = self._load_feature_module(feature_path)
            if not module:
                continue
            factory = getattr(module, "create_orchestrator", None)
            if not factory:
                continue
            base_name = feature_path.split(".")[-1]
            key = f"{prefix}{base_name}"
            orchestrators[key] = factory(container)
            log.info(f"FeatureDiscovery | orchestrator loaded key='{key}'")

    return orchestrators

discover_all()

Starts auto-discovery of all registered features.

For Telegram features: loads menu config, garbage states, Aiogram routers. For Redis features: loads RedisRouter and connects it to the dispatcher.

Source code in src/codex_bot/engine/discovery/service.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def discover_all(self) -> None:
    """
    Starts auto-discovery of all registered features.

    For Telegram features: loads menu config, garbage states, Aiogram routers.
    For Redis features: loads RedisRouter and connects it to the dispatcher.
    """
    for feature_path in self._features:
        self._discover_menu(feature_path)
        self._discover_garbage_states(feature_path)

    for feature_path in self._redis_features:
        self._discover_redis_handlers(feature_path)
        self._discover_garbage_states(feature_path)

get_menu_buttons(is_admin=None)

Returns menu button configurations for all Telegram features.

Parameters:

Name Type Description Default
is_admin bool | None

None — all buttons, True — only admin, False — only user.

 None

Returns:

Type Description
dict[str, dict[str, Any]]

Dictionary {feature_key: menu_config_dict}.
Source code in src/codex_bot/engine/discovery/service.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def get_menu_buttons(self, is_admin: bool | None = None) -> dict[str, dict[str, Any]]:
    """
    Returns menu button configurations for all Telegram features.

    Args:
        is_admin: None — all buttons, True — only admin, False — only user.

    Returns:
        Dictionary {feature_key: menu_config_dict}.
    """
    buttons: dict[str, dict[str, Any]] = {}
    for feature_path in self._features:
        btn = self._discover_menu(feature_path)
        if btn:
            if is_admin is not None and btn.get("is_admin", False) != is_admin:
                continue
            key = btn.get("key", feature_path)
            buttons[key] = btn
    return buttons

register_garbage_states(states)

Explicitly registers states as garbage (Garbage Collector).

Parameters:

Name Type Description Default
states Any

State, StatesGroup, string, or a list of them.

 required
Source code in src/codex_bot/engine/discovery/service.py 
195
196
197
198
199
200
201
202
def register_garbage_states(self, states: Any) -> None:
    """
    Explicitly registers states as garbage (Garbage Collector).

    Args:
        states: State, StatesGroup, string, or a list of them.
    """
    GarbageStateRegistry.register(states)

register_orchestrator(key, orchestrator)

Explicitly registers a feature orchestrator.

Parameters:

Name Type Description Default
key str

Feature key (e.g., "booking", "redis_notifications").

 required
orchestrator Any

Orchestrator instance.

 required
Source code in src/codex_bot/engine/discovery/service.py 
185
186
187
188
189
190
191
192
193
def register_orchestrator(self, key: str, orchestrator: Any) -> None:
    """
    Explicitly registers a feature orchestrator.

    Args:
        key: Feature key (e.g., "booking", "redis_notifications").
        orchestrator: Orchestrator instance.
    """
    self._explicit_orchestrators[key] = orchestrator

register_router(router)

Explicitly registers an Aiogram Router (without auto-discovery).

Parameters:

Name Type Description Default
router Router

Instance of aiogram.Router.

 required
Source code in src/codex_bot/engine/discovery/service.py 
176
177
178
179
180
181
182
183
def register_router(self, router: Router) -> None:
    """
    Explicitly registers an Aiogram Router (without auto-discovery).

    Args:
        router: Instance of aiogram.Router.
    """
    self._explicit_routers.append(router)