Skip to content
v0.1.3

Engine Module API Reference

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

bijux_cli.core.engine

Provides the core runtime engine for the Bijux CLI.

This module defines the Engine class, which is responsible for orchestrating the application's runtime environment after initial setup. Its key responsibilities include:

* Initializing and registering all default services with the Dependency
    Injection (DI) container.
* Discovering, loading, and registering all external plugins.
* Providing a central method for dispatching commands to plugins.
* Managing the graceful shutdown of services.

The engine acts as the bridge between the CLI command layer and the underlying services and plugins.

Engine 

Engine(
    di: Any = None,
    *,
    debug: bool = False,
    fmt: OutputFormat = JSON,
    quiet: bool = False,
)

Orchestrates the CLI's runtime services and plugin lifecycle.

Attributes:

  • _di (DIContainer) –

    The dependency injection container.

  • _debug (bool) –

    The debug mode flag.

  • _format (OutputFormat) –

    The default output format.

  • _quiet (bool) –

    The quiet mode flag.

  • _registry (RegistryProtocol) –

    The plugin registry service.

Initializes the engine and its core services.

This sets up the DI container, registers default services, and loads all discoverable plugins.

Parameters:

  • di (Any, default: None ) –

    An existing dependency injection container. If None, the global singleton instance is used. Defaults to None.

  • debug (bool, default: False ) –

    If True, enables debug mode for services.

  • fmt (OutputFormat, default: JSON ) –

    The default output format for services.

  • quiet (bool, default: False ) –

    If True, suppresses output from services.

Source code in src/bijux_cli/core/engine.py 
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
def __init__(
    self,
    di: Any = None,
    *,
    debug: bool = False,
    fmt: OutputFormat = OutputFormat.JSON,
    quiet: bool = False,
) -> None:
    """Initializes the engine and its core services.

    This sets up the DI container, registers default services, and loads
    all discoverable plugins.

    Args:
        di (Any, optional): An existing dependency injection container. If
            None, the global singleton instance is used. Defaults to None.
        debug (bool): If True, enables debug mode for services.
        fmt (OutputFormat): The default output format for services.
        quiet (bool): If True, suppresses output from services.
    """
    from bijux_cli.core.di import DIContainer

    self._di = di or DIContainer.current()
    self._debug = debug
    self._format = fmt
    self._quiet = quiet
    self._di.register(Observability, lambda: Observability(debug=debug))
    register_default_services(self._di, debug=debug, output_format=fmt, quiet=quiet)
    self._di.register(Engine, self)
    self._registry: RegistryProtocol = self._di.resolve(RegistryProtocol)
    self._register_plugins()

di property 

di: DIContainer

Read-only access to the DI container.

run_command async 

run_command(name: str, *args: Any, **kwargs: Any) -> Any

Executes a plugin's command with a configured timeout.

Parameters:

  • name (str) –

    The name of the command or plugin to run.

  • *args (Any, default: () ) –

    Positional arguments to pass to the plugin's execute method.

  • **kwargs (Any, default: {} ) –

    Keyword arguments to pass to the plugin's execute method.

Returns:

  • Any ( Any ) –

    The result of the command's execution.

Raises:

  • CommandError –

    If the plugin is not found, its execute method is invalid, or if it fails during execution.

Source code in src/bijux_cli/core/engine.py 
 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
async def run_command(self, name: str, *args: Any, **kwargs: Any) -> Any:
    """Executes a plugin's command with a configured timeout.

    Args:
        name (str): The name of the command or plugin to run.
        *args (Any): Positional arguments to pass to the plugin's `execute`
            method.
        **kwargs (Any): Keyword arguments to pass to the plugin's `execute`
            method.

    Returns:
        Any: The result of the command's execution.

    Raises:
        CommandError: If the plugin is not found, its `execute` method
            is invalid, or if it fails during execution.
    """
    plugin = self._registry.get(name)
    execute = getattr(plugin, "execute", None)
    if not callable(execute):
        raise CommandError(
            f"Plugin '{name}' has no callable 'execute' method.", http_status=404
        )
    if not inspect.iscoroutinefunction(execute):
        raise CommandError(
            f"Plugin '{name}' 'execute' is not async/coroutine.", http_status=400
        )
    try:
        return await asyncio.wait_for(execute(*args, **kwargs), self._timeout())
    except Exception as exc:  # pragma: no cover
        raise CommandError(f"Failed to run plugin '{name}': {exc}") from exc

run_repl async 

run_repl() -> None

Runs the interactive shell (REPL).

Note: This is a placeholder for future REPL integration.

Source code in src/bijux_cli/core/engine.py 
113
114
115
116
117
118
async def run_repl(self) -> None:
    """Runs the interactive shell (REPL).

    Note: This is a placeholder for future REPL integration.
    """
    pass

shutdown async 

shutdown() -> None

Gracefully shuts down the engine and all resolved services.

This method orchestrates the termination sequence for the application's runtime. It first attempts to flush any buffered command history to disk and then proceeds to shut down the main dependency injection container, which in turn cleans up all resolved services.

Returns:

  • None ( None ) –
Source code in src/bijux_cli/core/engine.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def shutdown(self) -> None:
    """Gracefully shuts down the engine and all resolved services.

    This method orchestrates the termination sequence for the application's
    runtime. It first attempts to flush any buffered command history to
    disk and then proceeds to shut down the main dependency injection
    container, which in turn cleans up all resolved services.

    Returns:
        None:
    """
    try:
        self._di.resolve(History).flush()
    except KeyError:
        pass
    finally:
        await self._di.shutdown()