commands

drunc.unified_shell.commands

Classes

Functions:

echo(obj, text, server, target_server, logger, target, severity, execute_along_path, execute_on_all_subsequent_children_in_path)

Log a message locally or send it to selected servers.

Without server mode, the message is logged locally. In server mode, it can be sent to the process manager, controller, or both.

Parameters:

Name	Type	Description	Default
obj	ProcessManagerContext	The unified shell context.	required
text	str	The log message text.	required
target_server	str	The server to target, or all servers when empty.	required
logger	int	The selected logger target.	required
severity	str	The log severity level.	required

Source code in drunc/unified_shell/commands.py

@click.command("echo")
@click.argument("text", required=True)
@click.option(
    "--target-server",
    type=str,
    default="",
    help="Server to use the log command on. Default value of '' will send the log message to all the servers, e.g. the process manager and the root controller.",
)
@click.option(
    "--server/--local",
    default=False,
    help="Send the message to the server via RPC (default: log locally only).",
)
@click.option(
    "--logger",
    type=click.Choice(["echo", "main"]),
    default="echo",
    callback=lambda context, parameter, value: LoggerTarget.Value(value.upper()),
    help="Which server-side logger to target when --server is used.",
)
@click.option(
    "-s",
    "--severity",
    type=str,
    default="INFO",
    help=(
        "Severity level of the log message (default INFO). Options: DEBUG, INFO, "
        "WARNING, ERROR, CRITICAL"
    ),
)
@click.option("--target", type=str, help="The session target to address", default="")
@click.option(
    "--execute-along-path/--dont-execute-along-path",
    is_flag=True,
    show_default=True,
    help="Execute the command along the session application path",
    default=False,
)
@click.option(
    "--execute-on-all-subsequent-children-in-path/--dont-execute-on-all-subsequent-children-in-path",
    is_flag=True,
    show_default=True,
    help="Execute the command on all subsequent children in the session application path",
    default=True,
)
@click.pass_obj
def echo(
    obj: ProcessManagerContext,
    text: str,
    server: bool,
    target_server: str,
    logger: int,
    target: str,
    severity: str,
    execute_along_path: bool,
    execute_on_all_subsequent_children_in_path: bool,
) -> None:
    """
    Log a message locally or send it to selected servers.

    Without server mode, the message is logged locally. In server mode, it can be sent
    to the process manager, controller, or both.

    Args:
        obj: The unified shell context.
        text: The log message text.
        target_server: The server to target, or all servers when empty.
        logger: The selected logger target.
        severity: The log severity level.
    """
    log = get_logger("unified_shell.echo")
    log.debug("Logging message to server(s)...")

    if not server:
        (log_echo if logger == LoggerTarget.ECHO else log).info(text)
        return

    if target_server in ["", "process_manager"]:
        obj.get_pm_driver().send_log(text=text, severity=severity, logger=logger)

    if target_server in ["", "controller"] and obj.has_driver("controller"):
        obj.get_controller_driver().send_log(
            text=text,
            severity=severity,
            target=target,
            logger=logger,
            execute_along_path=execute_along_path,
            execute_on_all_subsequent_children_in_path=execute_on_all_subsequent_children_in_path,
        )

session_injector(f)

Decorator to inject the session name into the command function.

This is used to wrap the relevant unified shell commands, as the unified shell is intended to only operate a single session at a time, and the session name is stored in the context object.

Parameters:

Name	Type	Description	Default
f	FC	The command function to wrap.	required

Returns:

Name	Type	Description
FC	FC	The wrapped command function with the session name injected.

Source code in drunc/unified_shell/commands.py

def session_injector(f: FC) -> FC:
    """
    Decorator to inject the session name into the command function.

    This is used to wrap the relevant unified shell commands, as the unified shell is
    intended to only operate a single session at a time, and the session name is stored
    in the context object.

    Args:
        f (FC): The command function to wrap.

    Returns:
        FC: The wrapped command function with the session name injected.

    Raises:
        None
    """

    @click.pass_context
    def wrapper(ctx: click.core.Context, *args: object, **kwargs: object) -> object:
        """
        Wrapper function to inject the session name into the command function.

        Args:
            ctx (click.core.Context): The click context object.
            *args (object): Positional arguments to pass to the command function.
            **kwargs (object): Keyword arguments to pass to the command function.

        Returns:
            object: The result of invoking the command function with the session name
                injected.

        Raises:
            None
        """
        kwargs["session"] = ctx.obj.session_name
        return ctx.invoke(f, *args, **kwargs)

    # The update_wrapper function is used to update the internal python methods for the
    # functions so that the click decorators can be used with the session_injector
    # decorator. This is necessary because the click decorators rely on the function
    # signature to determine the parameters that are passed to the function
    return cast(FC, update_wrapper(wrapper, f))

start_shell(ctx, obj)

Start an interactive shell session.

This command stops batch mode and enters an interactive shell state, allowing you to execute commands interactively.

Source code in drunc/unified_shell/commands.py

@click.command("start-shell")
@click.pass_obj
@click.pass_context
def start_shell(ctx: click.core.Context, obj: UnifiedShellContext) -> None:
    """
    Start an interactive shell session.

    This command stops batch mode and enters an interactive shell state,
    allowing you to execute commands interactively.
    """
    log = get_logger("unified_shell.start_shell")
    log_pm_cmd(obj)

    obj.running_mode = UnifiedShellMode.SEMIBATCH
    log.info("Switching to interactive mode...")

terminate(ctx, obj, width, fit)

Execute the process manager terminate command, but only do this for the current session

Source code in drunc/unified_shell/commands.py

@click.command("terminate")
@click.option(
    "-w",
    "--width",
    type=int,
    default=None,
    help="Table width. Default is automatically calculated",
)
@click.option(
    "--fit",
    is_flag=True,
    default=False,
    help="Restrict the table to the available terminal width.",
)
@click.pass_obj
@click.pass_context
def terminate(
    ctx: click.core.Context, obj: UnifiedShellContext, width: int | None, fit: bool
) -> None:
    """
    Execute the process manager terminate command, but only do this for the current
    session
    """

    log = get_logger("unified_shell.terminate")
    log_pm_cmd(obj)
    session_query = ProcessQuery(session=ctx.obj.session_name)
    log.info(f"Terminating session [green]{ctx.obj.session_name}[/]")
    result = obj.get_pm_driver().kill(session_query)
    if not result:
        return

    table = tabulate_process_instance_list(
        result, "Terminated process", False, width=width
    )

    table = format_table_width(obj, table, fit=fit)
    obj.print(table, soft_wrap=not fit)
    obj.delete_driver("controller")