Skip to content
v0.1.3

Entrypoints Module API Reference

This section documents the internals of the entrypoints module in Bijux CLI.

bijux_cli.services.plugins.entrypoints

Discovers and loads plugins distributed as Python packages.

This module provides the load_entrypoints function, which is responsible for finding and loading plugins that have been installed into the Python environment and registered under the bijux_cli.plugins entry point group. This enables a distributable plugin ecosystem where plugins can be managed via tools like pip.

load_entrypoints async 

load_entrypoints(
    di: DIContainer | None = None,
    registry: RegistryProtocol | None = None,
) -> None

Discovers, loads, and registers all entry point-based plugins.

This function iterates through all entry points in the 'bijux_cli.plugins' group. For each one, it attempts to load, instantiate, and register the plugin. It also performs an API version compatibility check and runs the plugin's startup hook if present.

Note

All exceptions during the loading or startup of a single plugin are caught, logged, and reported via telemetry. A failed plugin will be deregistered and will not prevent other plugins from loading.

Parameters:

  • di (DIContainer | None, default: None ) –

    The dependency injection container. If None, the current global container is used.

  • registry (RegistryProtocol | None, default: None ) –

    The plugin registry. If None, it is resolved from the DI container.

Returns:

  • None ( None ) –
Source code in src/bijux_cli/services/plugins/entrypoints.py 
 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
async def load_entrypoints(
    di: DIContainer | None = None,
    registry: RegistryProtocol | None = None,
) -> None:
    """Discovers, loads, and registers all entry point-based plugins.

    This function iterates through all entry points in the 'bijux_cli.plugins'
    group. For each one, it attempts to load, instantiate, and register the
    plugin. It also performs an API version compatibility check and runs the
    plugin's `startup` hook if present.

    Note:
        All exceptions during the loading or startup of a single plugin are
        caught, logged, and reported via telemetry. A failed plugin will be
        deregistered and will not prevent other plugins from loading.

    Args:
        di (DIContainer | None): The dependency injection container. If None,
            the current global container is used.
        registry (RegistryProtocol | None): The plugin registry. If None, it is
            resolved from the DI container.

    Returns:
        None:
    """
    import bijux_cli

    di = di or DIContainer.current()
    registry = registry or di.resolve(RegistryProtocol)

    obs = di.resolve(ObservabilityProtocol, None)
    tel = di.resolve(TelemetryProtocol, None)

    for ep in _iter_plugin_eps():
        try:
            plugin_class = ep.load()
            plugin = plugin_class()

            if not _compatible(plugin):
                raise RuntimeError(
                    f"Plugin '{ep.name}' requires API {getattr(plugin, 'requires_api_version', 'N/A')}, "
                    f"host is {bijux_cli.api_version}"
                )

            for tgt in (plugin_class, plugin):
                raw = getattr(tgt, "version", None)
                if raw is not None and not isinstance(raw, str):
                    tgt.version = str(raw)

            registry.register(ep.name, plugin, version=plugin.version)

            startup = getattr(plugin, "startup", None)
            if asyncio.iscoroutinefunction(startup):
                await startup(di)
            elif callable(startup):
                startup(di)

            if obs:
                obs.log("info", f"Loaded plugin '{ep.name}'")
            if tel:
                tel.event("entrypoint_plugin_loaded", {"name": ep.name})

        except Exception as exc:
            with contextlib.suppress(Exception):
                registry.deregister(ep.name)

            if obs:
                obs.log(
                    "error",
                    f"Failed to load plugin '{ep.name}'",
                    extra={"trace": traceback.format_exc(limit=5)},
                )
            if tel:
                tel.event(
                    "entrypoint_plugin_failed", {"name": ep.name, "error": str(exc)}
                )

            _LOG.debug("Skipped plugin %s: %s", ep.name, exc, exc_info=True)