commands

drunc.unified_shell.commands

Classes

Functions:

log_on_server(obj, text, target_server, severity, target, execute_along_path, execute_on_all_subsequent_children_in_path)

Log a message to the specified server.

This command allows you to send a log message to a specific server or to all servers in the system. You can specify the severity level of the log message.

Parameters:

Name	Type	Description	Default
obj	ProcessManagerContext	The context object containing session information.	required
text	str	The log message text.	required
target_server	str	The server to send the log message to. Default is '' (all servers).	required
severity	str	The severity level of the log message. Default is 'INFO'.	required

Returns:

Type	Description
None	None

Source code in drunc/unified_shell/commands.py

@click.command("log")
@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(
    "-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 log_on_server(
    obj: ProcessManagerContext,
    text: str,
    target_server: str,
    severity: str,
    target: str,
    execute_along_path: bool,
    execute_on_all_subsequent_children_in_path: bool,
) -> None:
    """
    Log a message to the specified server.

    This command allows you to send a log message to a specific server or to all servers
    in the system. You can specify the severity level of the log message.

    Args:
        obj (ProcessManagerContext): The context object containing session information.
        text (str): The log message text.
        target_server (str): The server to send the log message to. Default is '' (all servers).
        severity (str): The severity level of the log message. Default is 'INFO'.

    Returns:
        None

    Raises:
        None
    """
    log = get_logger("unified_shell.log_on_server")
    log.debug("Logging message to server(s)...")

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

    if target_server in ["", "controller"] and obj.has_driver("controller"):
        obj.get_controller_driver().log_on_server(
            text=text,
            severity=severity,
            target=target,
            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)

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.pass_obj
@click.pass_context
def terminate(ctx: click.core.Context, obj: UnifiedShellContext, width: int) -> 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

    obj.print(
        tabulate_process_instance_list(result, "Terminated process", False, width=width)
    )  # rich tables require console printing
    # As the session is now terminated, we can delete the controller driver, as it is no
    # longer needed.
    obj.delete_driver("controller")