call_command_wrapper

call_command_wrapper(*, cmd, log_path)

Call a command and write its stdout and stderr to files

PARAMETER	DESCRIPTION
cmd	TYPE: str
log_path	TYPE: str

Source code in fractal_server/runner/executors/call_command_wrapper.py

def call_command_wrapper(*, cmd: str, log_path: str) -> None:
    """
    Call a command and write its stdout and stderr to files

    Args:
        cmd:
        log_path:
    """
    try:
        validate_cmd(cmd)
    except ValueError as e:
        raise TaskExecutionError(f"Invalid command. Original error: {str(e)}")

    split_cmd = shlex.split(cmd)

    # Verify that task command is executable
    if shutil.which(split_cmd[0]) is None:
        msg = (
            f'Command "{split_cmd[0]}" is not valid. '
            "Hint: make sure that it is executable."
        )
        raise TaskExecutionError(msg)

    with open(log_path, "w") as fp_log:
        try:
            result = subprocess.run(  # nosec
                split_cmd,
                stderr=fp_log,
                stdout=fp_log,
            )
        except Exception as e:
            # This is likely unreachable
            raise e

    if result.returncode != 0:
        stderr = ""
        if os.path.isfile(log_path):
            with open(log_path) as fp_stderr:
                stderr = fp_stderr.read()
            stderr = placeholder_if_too_long(stderr)
        raise TaskExecutionError(
            f"Task failed with returncode={result.returncode}.\n"
            f"STDERR: {stderr}"
        )

placeholder_if_too_long(stderr)

Returns a placeholder if the string is too long

Source code in fractal_server/runner/executors/call_command_wrapper.py

def placeholder_if_too_long(stderr: str) -> str:
    """Returns a placeholder if the string is too long"""
    if len(stderr) > MAX_LEN_STDERR:
        return (
            f"Cannot display stderr of length {len(stderr)}. You can find the "
            "detailed logs by downloading the job-log folder."
        )
    return stderr