Skip to content

API reference

Reference for extension loading and startup entry points.

Extension loader

loader

Extension loading and initialization logic.

load_extensions() -> tuple[SetupFunctionList, WebserverFunctionList, StartupFunctionList, list[ExtensionTranslation]]

Load extensions from the extensions directory.

Returns:

Type Description
SetupFunctionList

A tuple containing:

WebserverFunctionList
  • bot_functions: List of (setup_function, config) tuples for bot setup
StartupFunctionList
  • back_functions: List of (setup_webserver_function, config) tuples for backend setup
list[ExtensionTranslation]
  • startup_functions: List of (on_startup_function, config) tuples for startup hooks
tuple[SetupFunctionList, WebserverFunctionList, StartupFunctionList, list[ExtensionTranslation]]
  • translations: List of loaded extension translations
Source code in src/startup/loader.py
 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
def load_extensions() -> tuple[
    SetupFunctionList,
    WebserverFunctionList,
    StartupFunctionList,
    list[ExtensionTranslation],
]:
    """Load extensions from the extensions directory.

    Returns:
        A tuple containing:
        - bot_functions: List of (setup_function, config) tuples for bot setup
        - back_functions: List of (setup_webserver_function, config) tuples for backend setup
        - startup_functions: List of (on_startup_function, config) tuples for startup hooks
        - translations: List of loaded extension translations

    """
    from src.utils import validate_module  # noqa: PLC0415

    bot_functions: SetupFunctionList = []
    back_functions: WebserverFunctionList = []
    startup_functions: StartupFunctionList = []
    translations: list[ExtensionTranslation] = []

    for _extension in iglob("src/extensions/*"):
        extension_path = Path(_extension)
        name = extension_path.name

        # Skip special files
        if name.endswith(("_", "_/", ".py")):
            continue

        # Check if extension is enabled in config
        _, its_config = config.get_extension(name, {})
        if its_config and not its_config.get("enabled"):
            continue

        # Try to import the extension module
        try:
            module: ModuleType = importlib.import_module(f"src.extensions.{name}")
        except ImportError as e:
            logger.error(f"Failed to import extension {name}")
            logger.debug("", exc_info=e)
            continue

        # Get the extension's config (from config file or module's default)
        module_default: dict[str, Any] = getattr(module, "default", {})
        its_config = its_config or module_default
        if not its_config.get("enabled"):
            del module
            continue

        logger.info(f"Loading extension {name}")

        # Load translations if available
        translation_path = find_translation_file(extension_path, name)
        if translation_path is not None:
            try:
                translation = i18n.load_translation(str(translation_path))
                translations.append(translation)

                # Store translation for later use
                if translation.strings:
                    its_config["translations"] = translation.strings
            except yaml.YAMLError as e:
                logger.error(f"Error loading translation {translation_path}: {e}")
        else:
            logger.warning(f"No translation found for extension {name}")

        # Validate the module structure
        validate_module(module, its_config)

        # Register extension functions
        # Type checkers can't infer the exact types from hasattr/callable checks,
        # so we use type: ignore to suppress false positives
        if hasattr(module, "setup") and callable(module.setup):
            bot_functions.append((module.setup, its_config))  # pyright: ignore[reportArgumentType]

        if hasattr(module, "setup_webserver") and callable(module.setup_webserver):
            back_functions.append((module.setup_webserver, its_config))  # pyright: ignore[reportArgumentType]

        if hasattr(module, "on_startup") and callable(module.on_startup):
            startup_functions.append((module.on_startup, its_config))  # pyright: ignore[reportArgumentType]

    return bot_functions, back_functions, startup_functions, translations

find_translation_file(extension_path: Path, extension_name: str) -> Path | None

Find translation file for an extension using centralized path resolution.

Searches in the following order: 1. extension_path/translations.yml 2. extension_path/translations.yaml 3. src/translations/{extension_name}.yml 4. src/translations/{extension_name}.yaml

Parameters:

Name Type Description Default
extension_path Path

Path to the extension directory

required
extension_name str

Name of the extension

required

Returns:

Type Description
Path | None

Path to the translation file if found, None otherwise

Source code in src/startup/loader.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
def find_translation_file(extension_path: Path, extension_name: str) -> Path | None:
    """Find translation file for an extension using centralized path resolution.

    Searches in the following order:
    1. extension_path/translations.yml
    2. extension_path/translations.yaml
    3. src/translations/{extension_name}.yml
    4. src/translations/{extension_name}.yaml

    Args:
        extension_path: Path to the extension directory
        extension_name: Name of the extension

    Returns:
        Path to the translation file if found, None otherwise

    """
    candidates = [
        extension_path / "translations.yml",
        extension_path / "translations.yaml",
        Path(__file__).parent.parent / "translations" / f"{extension_name}.yml",
        Path(__file__).parent.parent / "translations" / f"{extension_name}.yaml",
    ]

    for candidate in candidates:
        if candidate.exists():
            return candidate

    return None

Extension utilities

extensions

validate_module(module: ModuleType, config: dict[str, object] | None = None) -> None

Validate the module to ensure it has the required functions and attributes to be loaded as an extension.

Parameters:

Name Type Description Default
module ModuleType

The module to validate

required
config dict[str, object] | None

The configuration to validate against (currently unused)

None

Raises:

Type Description
AssertionError

If the module doesn't meet extension requirements

Source code in src/utils/extensions.py
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
43
44
45
46
47
48
49
50
51
def validate_module(module: ModuleType, config: dict[str, object] | None = None) -> None:  # noqa: ARG001
    """Validate the module to ensure it has the required functions and attributes to be loaded as an extension.

    Args:
        module: The module to validate
        config: The configuration to validate against (currently unused)

    Raises:
        AssertionError: If the module doesn't meet extension requirements

    """
    if hasattr(module, "setup"):
        assert isinstance(module.setup, SetupFunction), (
            f"Extension {module.__name__} has an invalid setup function signature"
        )

    if hasattr(module, "setup_webserver"):
        assert isinstance(module.setup_webserver, SetupWebserverFunction), (
            f"Extension {module.__name__} has an invalid setup_webserver function signature"
        )

    assert hasattr(module, "setup_webserver") or hasattr(
        module,
        "setup",
    ), f"Extension {module.__name__} does not have a setup or setup_webserver function"

    if hasattr(module, "on_startup"):
        assert isinstance(module.on_startup, StartupFunction), (
            f"Extension {module.__name__} has an invalid on_startup function signature"
        )

    assert hasattr(module, "default"), f"Extension {module.__name__} does not have a default configuration"
    assert isinstance(
        module.default,
        dict,
    ), f"Extension {module.__name__} has a default configuration of type {type(module.default)} instead of dict"
    assert "enabled" in module.default, (
        f"Extension {module.__name__} does not have an enabled key in its default configuration"
    )

unzip_extensions() -> None

Source code in src/utils/extensions.py
54
55
56
57
58
59
def unzip_extensions() -> None:
    for file in iglob("src/extensions/*.zip"):
        with zipfile.ZipFile(file, "r") as zip_ref:
            zip_ref.extractall("src/extensions")
            os.remove(file)
            logger.info(f"Extracted {file}")

Startup orchestration

start

Main entry point for starting the bot and backend server.

This module provides the top-level orchestration for the startup process. Most of the actual implementation has been moved to the src.startup package for better organization and maintainability.

start(run_bot: bool | None = None, run_backend: bool | None = None) -> None async

Start the bot and/or backend server based on configuration.

Parameters:

Name Type Description Default
run_bot bool | None

Whether to start the bot (defaults to config.use.bot)

None
run_backend bool | None

Whether to start the backend server (defaults to config.use.backend)

None
Source code in src/start.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
106
107
108
109
110
111
112
113
114
115
async def start(run_bot: bool | None = None, run_backend: bool | None = None) -> None:
    """Start the bot and/or backend server based on configuration.

    Args:
        run_bot: Whether to start the bot (defaults to config.use.bot)
        run_backend: Whether to start the backend server (defaults to config.use.backend)

    """
    if not config.bot.token:
        logger.critical("No bot token provided in config, exiting...")
        return

    if config.db.enabled:
        from src.database.config import init as init_db  # noqa: PLC0415

        logger.info("Initializing database...")
        await init_db()

    unzip_extensions()

    run_bot = run_bot if run_bot is not None else config.use.bot
    run_backend = run_backend if run_backend is not None else config.use.backend

    bot_functions, back_functions, startup_functions, translations = load_extensions()

    start_bot_extensions = bool(bot_functions and run_bot)
    start_backend_server = bool(back_functions and run_backend)

    if not start_bot_extensions and not start_backend_server:
        logger.error("Nothing to start, exiting...")
        return

    app = None
    bot = create_bot(config.bot)
    if start_bot_extensions:
        setup_bot(bot, bot_functions, translations, config.bot)
    if start_backend_server:
        app = create_backend_app()
        setup_backend_extensions(app, bot, back_functions)

    if startup_functions:
        await run_startup_functions(startup_functions, app, bot)

    if start_bot_extensions and start_backend_server:
        if config.bot.rest:
            logger.critical(
                "REST bot mode and the Botkit backend cannot run together in one process. "
                "Disable bot.rest or use.backend."
            )
            return
        if app is None:
            logger.error("Backend app was not initialized, exiting...")
            return
        await run_bot_and_backend(bot, app, config.bot)
    elif start_bot_extensions:
        await start_bot(bot, config.bot.token, config.bot.rest, config.bot.public_key)
    elif app is None:
        logger.error("Backend app was not initialized, exiting...")
    else:
        await run_backend_only(app, bot, config.bot.token, config.backend)