Api Module API Reference¶

def load_plugin(self, path: str | Path) -> None:
    """Loads or reloads a plugin module from a file path.

    This method dynamically loads the specified plugin file, initializes it,
    and registers it with the CLI system. If the plugin is already loaded,
    it is reloaded.

    Args:
        path (str | Path): The filesystem path to the plugin's Python file.

    Raises:
        BijuxError: If plugin loading, initialization, or registration fails.
    """
    from bijux_cli.__version__ import __version__
    from bijux_cli.services.plugins import load_plugin as _load_plugin

    p = Path(path).expanduser().resolve()
    module_name = f"bijux_plugin_{p.stem}"

    try:
        if module_name in sys.modules:
            importlib.reload(sys.modules[module_name])

        plugin = _load_plugin(p, module_name)
        plugin.startup(self._engine.di)

        exists = bool(
            self._await_maybe(self._registry.has(p.stem), want_result=True)
        )
        if exists:
            self._await_maybe(cast(Any, self._registry).deregister(p.stem))

        self._await_maybe(
            cast(Any, self._registry).register(
                p.stem, plugin, alias=str(__version__)
            )
        )
        self._obs.log("info", "Loaded plugin", extra={"path": str(p)})
        self._schedule_event("api.plugin_loaded", {"path": str(p)})

    except Exception as exc:
        self._schedule_event(
            "api.plugin_load.error", {"path": str(p), "error": str(exc)}
        )
        raise BijuxError(
            f"Failed to load plugin {p}: {exc}", http_status=500
        ) from exc

def register(self, name: str, callback: Callable[..., Any]) -> None:
    """Registers or replaces a Python callable as a CLI command.

    The provided callable is wrapped to handle both synchronous and
    asynchronous functions automatically.

    Args:
        name (str): The command name to register.
        callback (Callable[..., Any]): The Python function to be executed
            when the command is run.

    Raises:
        BijuxError: If the command name is already in use or another
            registration error occurs.
    """

    class _Wrapper:
        """Wraps a user-provided callable to be executed asynchronously."""

        def __init__(self, cb: Callable[..., Any]) -> None:
            """Initializes the wrapper.

            Args:
                cb (Callable[..., Any]): The callable to wrap.
            """
            self._cb = cb

        async def execute(self, *args: Any, **kwargs: Any) -> Any:
            """Execute the wrapped callable, awaiting if it's a coroutine.

            Args:
                *args (Any): Positional arguments to pass to the callable.
                **kwargs (Any): Keyword arguments to pass to the callable.

            Returns:
                Any: The result of the callable execution.
            """
            if asyncio.iscoroutinefunction(self._cb):
                return await self._cb(*args, **kwargs)
            return self._cb(*args, **kwargs)

    try:
        exists = bool(self._await_maybe(self._registry.has(name), want_result=True))
        if exists:
            maybe = cast(Any, self._registry.deregister(name))
            self._await_maybe(maybe)
        maybe2 = cast(Any, self._registry.register(name, _Wrapper(callback)))
        self._await_maybe(maybe2)

        self._obs.log("info", "Registered command", extra={"name": name})
        self._schedule_event("api.register", {"name": name})
    except ServiceError as exc:
        self._schedule_event(
            "api.register.error", {"name": name, "error": str(exc)}
        )
        raise BijuxError(
            f"Could not register command {name}: {exc}", http_status=500
        ) from exc

async def run_async(
    self,
    name: str,
    *args: Any,
    quiet: bool = False,
    verbose: bool = False,
    fmt: str = "json",
    pretty: bool = True,
    debug: bool = False,
    **kwargs: Any,
) -> Any:
    """Runs a command asynchronously with validation.

    This method performs validation of flags and environment variables
    before dispatching the command to the internal engine for execution.

    Args:
        name (str): The name of the command to execute.
        *args (Any): Positional arguments for the command.
        quiet (bool): If True, suppresses output.
        verbose (bool): If True, enables verbose output.
        fmt (str): The output format ("json" or "yaml").
        pretty (bool): If True, formats the output for readability.
        debug (bool): If True, enables debug mode.
        **kwargs (Any): Additional keyword arguments to pass to the command.

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

    Raises:
        BijuxError: For invalid flags, unsupported formats, or internal
            execution errors.
    """
    try:
        fmt = fmt.lower()
        if fmt not in ("json", "yaml"):
            raise BijuxError("Unsupported format", http_status=400)

        if quiet and (verbose or debug):
            raise BijuxError(
                "--quiet cannot be combined with --verbose/--debug", http_status=400
            )

        validate_common_flags(fmt, name, quiet, verbose or debug)

        for k, v in os.environ.items():
            if k in IGNORE:
                continue
            if not v.isascii():
                raise BijuxError(
                    "Non-ASCII characters in environment", http_status=400
                )

        result = await self._engine.run_command(name, *args, **kwargs)
        self._schedule_event("api.run", {"name": name})
        return result

    except CommandError as exc:
        self._schedule_event("api.run.error", {"name": name, "error": str(exc)})
        raise BijuxError(
            f"Failed to run command {name}: {exc}", http_status=500
        ) from exc

    except ServiceError as exc:
        self._schedule_event("api.run.error", {"name": name, "error": str(exc)})
        raise BijuxError(
            f"Failed to run command {name}: {exc}", http_status=500
        ) from exc

    except BijuxError:
        raise

    except Exception as exc:
        self._schedule_event("api.run.error", {"name": name, "error": str(exc)})
        raise BijuxError(
            f"Failed to run command {name}: {exc}", http_status=500
        ) from exc

def run_sync(
    self,
    name: str,
    *args: Any,
    quiet: bool = False,
    verbose: bool = False,
    fmt: str = "json",
    pretty: bool = True,
    debug: bool = False,
    **kwargs: Any,
) -> Any:
    """Runs a command synchronously.

    This method is a blocking wrapper around the asynchronous `run_async`
    method. It manages the asyncio event loop to provide a simple,
    synchronous interface.

    Args:
        name (str): The name of the command to run.
        *args (Any): Positional arguments for the command.
        quiet (bool): If True, suppresses output.
        verbose (bool): If True, enables verbose logging.
        fmt (str): The output format ("json" or "yaml").
        pretty (bool): If True, formats the output for readability.
        debug (bool): If True, enables debug mode.
        **kwargs (Any): Additional keyword arguments to pass to the command.

    Returns:
        Any: The result of the command's execution.
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(
            self.run_async(
                name,
                *args,
                quiet=quiet,
                verbose=verbose,
                fmt=fmt,
                pretty=pretty,
                debug=debug,
                **kwargs,
            )
        )
    else:
        if not hasattr(loop, "run_until_complete"):
            raise RuntimeError("Cannot call run_sync from a running event loop")
        coro = self.run_async(
            name,
            *args,
            quiet=quiet,
            verbose=verbose,
            fmt=fmt,
            pretty=pretty,
            debug=debug,
            **kwargs,
        )
        try:
            return loop.run_until_complete(coro)
        finally:
            with suppress(Exception):
                if hasattr(coro, "close"):
                    coro.close()