tool_run

`tool_run` #

run_tool — subprocess.run for a recipe: the child's output streams into exec.log.

Both of the child's streams are pumped, as they arrive, into the ambient workspace's exec.log (see exec_log); only a capturing caller's streams are also buffered. Outside any workspace the call is a plain subprocess.run.

`run_tool(owner, argv, *, cwd, env, timeout, check, capture_output)` #

Run argv in cwd the way subprocess.run does, logging for owner.

Parameters:

Name	Type	Description	Default
`owner`	`object`	The value running the tool; its workspace decides where the log is.	required
`argv`	`Sequence[str \| PathLike[str]]`	The command line.	required
`cwd`	`PathLike[str]`	The working directory.	required
`env`	`Mapping[str, str] \| None`	The child's environment, or the parent's when `None`.	required
`timeout`	`float \| None`	Seconds before the child is killed.	required
`check`	`bool`	Whether a non-zero exit raises.	required
`capture_output`	`bool`	Whether stdout and stderr come back on the result.	required

Raises:

Type	Description
`CalledProcessError`	If `check` and the child exits non-zero.
`TimeoutExpired`	If the child outlives `timeout`.

Source code in capturegraph-lib/capturegraph/recipes/values/tool_run.py

def run_tool(
    owner: object,
    argv: Sequence[str | os.PathLike[str]],
    *,
    cwd: os.PathLike[str],
    env: Mapping[str, str] | None,
    timeout: float | None,
    check: bool,
    capture_output: bool,
) -> subprocess.CompletedProcess[bytes]:
    """Run ``argv`` in ``cwd`` the way ``subprocess.run`` does, logging for ``owner``.

    Args:
        owner: The value running the tool; its workspace decides where the log is.
        argv: The command line.
        cwd: The working directory.
        env: The child's environment, or the parent's when ``None``.
        timeout: Seconds before the child is killed.
        check: Whether a non-zero exit raises.
        capture_output: Whether stdout and stderr come back on the result.

    Raises:
        subprocess.CalledProcessError: If ``check`` and the child exits non-zero.
        subprocess.TimeoutExpired: If the child outlives ``timeout``.
    """
    command = [os.fspath(argument) for argument in argv]
    log = exec_log(owner)
    if log is None:
        return subprocess.run(
            command,
            cwd=cwd,
            env=env,
            timeout=timeout,
            check=check,
            capture_output=capture_output,
        )
    with open(log, "ab") as sink:
        sink.write(f"$ {shlex.join(command)}\n".encode())
        sink.flush()
        return _run_logged(command, sink, cwd, env, timeout, check, capture_output)

tool_run

tool_run #

run_tool(owner, argv, *, cwd, env, timeout, check, capture_output) #

`tool_run` #

`run_tool(owner, argv, *, cwd, env, timeout, check, capture_output)` #