Skip to content

engine.discovery — Feature discovery

FeatureDiscoveryService

FeatureDiscoveryService

Automated orchestration for framework-wide feature registration.

Operates on a 'convention over configuration' principle to dynamically locate and initialize business features. Manages the discovery of Aiogram routers, Redis stream handlers, UI menu configurations, andч FSM garbage collection states.

Source code in src/codex_bot/engine/discovery/service.py 
 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
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
class FeatureDiscoveryService:
    """Automated orchestration for framework-wide feature registration.

    Operates on a 'convention over configuration' principle to dynamically
    locate and initialize business features. Manages the discovery of
    Aiogram routers, Redis stream handlers, UI menu configurations, andч
    FSM garbage collection states.
    """

    def __init__(
        self,
        module_prefix: str = "features",
        installed_features: list[str] | None = None,
        installed_redis_features: list[str] | None = None,
        redis_dispatcher: BotStreamDispatcher | None = None,
    ) -> None:
        """
        Initializes the discovery service.

        Args:
            module_prefix: Base package for features (default: "features").
            installed_features: Simple folder names of Telegram features.
            installed_redis_features: Simple folder names of Redis features.
            redis_dispatcher: Dispatcher for registering Redis handlers.
        """
        self._prefix = module_prefix
        self._features = installed_features or []
        self._redis_features = installed_redis_features or []
        self._redis_dispatcher = redis_dispatcher

        # Explicitly registered objects (fallback for non-dynamic environments)
        self._explicit_routers: list[Router] = []
        self._explicit_orchestrators: dict[str, Any] = {}

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

        For Telegram features: loads menu config, garbage states, Aiogram routers.
        For Redis features: loads RedisRouter and connects it to the dispatcher.
        """
        for name in self._features:
            module = self._load_module(name, "telegram")
            if module:
                self._register_menu(module, name)
                self._register_garbage(module)

        for name in self._redis_features:
            module = self._load_module(name, "redis")
            if module:
                self._register_redis_handlers(module, name)
                self._register_garbage(module)

    def discover_models(self, project_name: str | None = None) -> None:
        """
        Dynamically imports the 'models' module for each installed feature.
        This ensures all SQLAlchemy models are registered with Base.metadata
        for Alembic migration discovery.

        Args:
            project_name: The root package name of the project (e.g., "my_bot").
        """
        prefix = f"{project_name}." if project_name else ""

        # 1. Discover models in Telegram features
        for name in self._features:
            module_path = f"{prefix}{self._prefix}.telegram.{name}.models"
            try:
                importlib.import_module(module_path)
                log.debug(f"FeatureDiscovery | Imported models from: {module_path}")
            except ImportError as e:
                # If module is missing, it's fine — some features might not have models
                if name in str(e):
                    log.debug(f"FeatureDiscovery | No models found in: {module_path}")
                else:
                    log.warning(f"FeatureDiscovery | Error importing models from {module_path}: {e}")

        # 2. Discover models in Redis features
        for name in self._redis_features:
            module_path = f"{prefix}{self._prefix}.redis.{name}.models"
            try:
                importlib.import_module(module_path)
                log.debug(f"FeatureDiscovery | Imported models from: {module_path}")
            except ImportError:
                pass

    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: list[tuple[list[str], Literal["telegram", "redis"], str]] = [
            (self._features, "telegram", ""),
            (self._redis_features, "redis", "redis_"),
        ]

        for names, f_type, key_prefix in configs:
            for name in names:
                module = self._load_module(name, f_type)
                if not module:
                    continue

                factory = getattr(module, "create_orchestrator", None)
                if factory and callable(factory):
                    key = f"{key_prefix}{name}"
                    orchestrators[key] = factory(container)
                    log.info(f"FeatureDiscovery | Orchestrator loaded: {key}")

        return orchestrators

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

        Expects routers to be named 'router' in '{name}.handlers.handlers'.

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

        for name in self._features:
            path = f"{self._prefix}.telegram.{name}.handlers"
            try:
                module = importlib.import_module(path)
                router = getattr(module, "router", None)
                if isinstance(router, Router):
                    routers.append(router)
                    log.info(f"FeatureDiscovery | Router loaded: {name}")
            except ImportError:
                log.debug(f"FeatureDiscovery | No handlers found for {name} at {path}")
                continue

        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 name in self._features:
            module = self._load_module(name, "telegram")
            if module:
                btn = getattr(module, "MENU_CONFIG", None)
                if isinstance(btn, dict):
                    if is_admin is not None and btn.get("is_admin", False) != is_admin:
                        continue
                    buttons[btn.get("key", name)] = btn
        return buttons

    # =========================================================================
    # Explicit Registration (Fallback)
    # =========================================================================

    def register_router(self, router: Router) -> None:
        """Explicitly registers an Aiogram Router."""
        self._explicit_routers.append(router)

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

    def register_garbage_states(self, states: Any) -> None:
        """Explicitly registers states for the Garbage Collector."""
        GarbageStateRegistry.register(states)

    # =========================================================================
    # Internal Logic
    # =========================================================================

    def _load_module(self, name: str, f_type: Literal["telegram", "redis"]) -> ModuleType | None:
        """Loads the feature_setting module following the convention."""
        path = f"{self._prefix}.{f_type}.{name}.feature_setting"
        try:
            return importlib.import_module(path)
        except ImportError:
            return None

    def _register_menu(self, module: ModuleType, name: str) -> None:
        """Internal helper to extract menu config."""
        pass

    def _register_garbage(self, module: ModuleType) -> None:
        """Registers states for the Garbage Collector."""
        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 _register_redis_handlers(self, module: ModuleType, name: str) -> None:
        """Connects RedisRouter to the dispatcher."""
        if not self._redis_dispatcher:
            return

        path = f"{self._prefix}.redis.{name}.handlers"
        try:
            handler_module = importlib.import_module(path)
            router = getattr(handler_module, "redis_router", None)
            if isinstance(router, StreamRouter):
                self._redis_dispatcher.include_router(router)
        except ImportError:
            pass

Functions

__init__(module_prefix='features', installed_features=None, installed_redis_features=None, redis_dispatcher=None)

Initializes the discovery service.

Parameters:

Name Type Description Default
module_prefix str

Base package for features (default: "features").

 'features'
installed_features list[str] | None

Simple folder names of Telegram features.

 None
installed_redis_features list[str] | None

Simple folder names of Redis features.

 None
redis_dispatcher BotStreamDispatcher | None

Dispatcher for registering Redis handlers.

 None
Source code in src/codex_bot/engine/discovery/service.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    module_prefix: str = "features",
    installed_features: list[str] | None = None,
    installed_redis_features: list[str] | None = None,
    redis_dispatcher: BotStreamDispatcher | None = None,
) -> None:
    """
    Initializes the discovery service.

    Args:
        module_prefix: Base package for features (default: "features").
        installed_features: Simple folder names of Telegram features.
        installed_redis_features: Simple folder names of Redis features.
        redis_dispatcher: Dispatcher for registering Redis handlers.
    """
    self._prefix = module_prefix
    self._features = installed_features or []
    self._redis_features = installed_redis_features or []
    self._redis_dispatcher = redis_dispatcher

    # Explicitly registered objects (fallback for non-dynamic environments)
    self._explicit_routers: list[Router] = []
    self._explicit_orchestrators: dict[str, Any] = {}

collect_aiogram_routers()

Collects Aiogram Routers from all Telegram features.

Expects routers to be named 'router' in '{name}.handlers.handlers'.

Returns:

Type Description
list[Router]

List of Routers to be included in the main dispatcher.
Source code in src/codex_bot/engine/discovery/service.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def collect_aiogram_routers(self) -> list[Router]:
    """
    Collects Aiogram Routers from all Telegram features.

    Expects routers to be named 'router' in '{name}.handlers.handlers'.

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

    for name in self._features:
        path = f"{self._prefix}.telegram.{name}.handlers"
        try:
            module = importlib.import_module(path)
            router = getattr(module, "router", None)
            if isinstance(router, Router):
                routers.append(router)
                log.info(f"FeatureDiscovery | Router loaded: {name}")
        except ImportError:
            log.debug(f"FeatureDiscovery | No handlers found for {name} at {path}")
            continue

    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 
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
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: list[tuple[list[str], Literal["telegram", "redis"], str]] = [
        (self._features, "telegram", ""),
        (self._redis_features, "redis", "redis_"),
    ]

    for names, f_type, key_prefix in configs:
        for name in names:
            module = self._load_module(name, f_type)
            if not module:
                continue

            factory = getattr(module, "create_orchestrator", None)
            if factory and callable(factory):
                key = f"{key_prefix}{name}"
                orchestrators[key] = factory(container)
                log.info(f"FeatureDiscovery | Orchestrator loaded: {key}")

    return orchestrators

discover_all()

Starts auto-discovery of all registered features based on convention.

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 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def discover_all(self) -> None:
    """
    Starts auto-discovery of all registered features based on convention.

    For Telegram features: loads menu config, garbage states, Aiogram routers.
    For Redis features: loads RedisRouter and connects it to the dispatcher.
    """
    for name in self._features:
        module = self._load_module(name, "telegram")
        if module:
            self._register_menu(module, name)
            self._register_garbage(module)

    for name in self._redis_features:
        module = self._load_module(name, "redis")
        if module:
            self._register_redis_handlers(module, name)
            self._register_garbage(module)

discover_models(project_name=None)

Dynamically imports the 'models' module for each installed feature. This ensures all SQLAlchemy models are registered with Base.metadata for Alembic migration discovery.

Parameters:

Name Type Description Default
project_name str | None

The root package name of the project (e.g., "my_bot").

 None
Source code in src/codex_bot/engine/discovery/service.py 
 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
def discover_models(self, project_name: str | None = None) -> None:
    """
    Dynamically imports the 'models' module for each installed feature.
    This ensures all SQLAlchemy models are registered with Base.metadata
    for Alembic migration discovery.

    Args:
        project_name: The root package name of the project (e.g., "my_bot").
    """
    prefix = f"{project_name}." if project_name else ""

    # 1. Discover models in Telegram features
    for name in self._features:
        module_path = f"{prefix}{self._prefix}.telegram.{name}.models"
        try:
            importlib.import_module(module_path)
            log.debug(f"FeatureDiscovery | Imported models from: {module_path}")
        except ImportError as e:
            # If module is missing, it's fine — some features might not have models
            if name in str(e):
                log.debug(f"FeatureDiscovery | No models found in: {module_path}")
            else:
                log.warning(f"FeatureDiscovery | Error importing models from {module_path}: {e}")

    # 2. Discover models in Redis features
    for name in self._redis_features:
        module_path = f"{prefix}{self._prefix}.redis.{name}.models"
        try:
            importlib.import_module(module_path)
            log.debug(f"FeatureDiscovery | Imported models from: {module_path}")
        except ImportError:
            pass

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 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
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 name in self._features:
        module = self._load_module(name, "telegram")
        if module:
            btn = getattr(module, "MENU_CONFIG", None)
            if isinstance(btn, dict):
                if is_admin is not None and btn.get("is_admin", False) != is_admin:
                    continue
                buttons[btn.get("key", name)] = btn
    return buttons

register_garbage_states(states)

Explicitly registers states for the Garbage Collector.

Source code in src/codex_bot/engine/discovery/service.py 
201
202
203
def register_garbage_states(self, states: Any) -> None:
    """Explicitly registers states for the Garbage Collector."""
    GarbageStateRegistry.register(states)

register_orchestrator(key, orchestrator)

Explicitly registers a feature orchestrator.

Source code in src/codex_bot/engine/discovery/service.py 
197
198
199
def register_orchestrator(self, key: str, orchestrator: Any) -> None:
    """Explicitly registers a feature orchestrator."""
    self._explicit_orchestrators[key] = orchestrator

register_router(router)

Explicitly registers an Aiogram Router.

Source code in src/codex_bot/engine/discovery/service.py 
193
194
195
def register_router(self, router: Router) -> None:
    """Explicitly registers an Aiogram Router."""
    self._explicit_routers.append(router)