Function _spawn_proc

bbot/core/helpers/command.py:163–205

Spawns an asynchronous subprocess. Prepares the command and associated keyword arguments. If the `input` argument is provided, it checks to ensure that the `stdin` argument is not also provided. Once prepared, it creates and returns the subprocess. If the command executable is not found

(self, *command, **kwargs)

async def _spawn_proc(self, *command, **kwargs):
    """Spawns an asynchronous subprocess.

    Prepares the command and associated keyword arguments. If the `input` argument is provided,
    it checks to ensure that the `stdin` argument is not also provided. Once prepared, it creates
    and returns the subprocess. If the command executable is not found, it logs a warning and traceback.

    Args:
        *command (str): The command to run as separate arguments.
        **kwargs (dict): Additional keyword arguments for the subprocess.

    Raises:
        ValueError: If both stdin and input arguments are provided.

    Returns:
        tuple: A tuple containing the created process (or None if creation failed), the input (or None if not provided),
               and the prepared command (or None if subprocess creation failed).

    Examples:
        >>> _spawn_proc("ls", "-l", input="data")
        (<Process ...>, "data", ["ls", "-l"])
    """
    try:
        command, kwargs = self._prepare_command_kwargs(command, kwargs)
    except SubprocessError as e:
        command_str = " ".join([str(s) for s in command])
        log.warning(f"Error running command: '{command_str}': {e}")
        log.trace(traceback.format_exc())
        return None, None, None
    _input = kwargs.pop("input", None)
    if _input is not None:
        if kwargs.get("stdin") is not None:
            raise ValueError("stdin and input arguments may not both be used.")
        kwargs["stdin"] = asyncio.subprocess.PIPE

    log.hugeverbose(f"run: {' '.join(command)}")
    try:
        proc = await asyncio.create_subprocess_exec(*command, **kwargs)
        return proc, _input, command
    except FileNotFoundError as e:
        log.warning(f"{e} - missing executable?")
        log.trace(traceback.format_exc())
    return None, None, None


async def _write_proc_line(proc, chunk):