stepup.core.extapi

Utilities for developers building StepUp extension packages. These functions are not intended for use in plan.py files.

stepup.core.extapi.run_subprocess(cmd, *, workdir='.', stdin=None, shell=False, check=True, text=None)

Run a subprocess and record it for archival purposes.

This is the convenience wrapper for the case where an extension step wraps an executable. The invocation, its return code, and its captured stdout/stderr are recorded via record_subprocess (subject to the STEPUP_MAX_OUTPUT_SIZE cap).

Parameters:

• cmd (str) –

  The command line, as a single shell-quoted string. When shell=False (the default), cmd is split with shlex.split and executed directly (no shell), so shell features (pipes, redirections, …) are not available. When shell=True, cmd is passed as-is to the system shell, which enables shell features. As an exception, leading VAR=value assignments are extracted and applied to the subprocess environment, even when shell=False. In either case, the caller is responsible for proper quoting.

• workdir (StrPath, default: '.' ) –

  The working directory of the subprocess as a path or string, relative to the step’s own working directory. It is passed to subprocess.run as cwd.

• stdin (str | bytes | None, default: None ) –

  Standard input fed to the subprocess, or None. A str is passed to subprocess.run as-is and implies text=True. bytes are passed as-is as well and imply text=False. Inconsistent combinations (e.g. stdin is bytes but text=True) raise a TypeError. The value is forwarded to record_subprocess, which stores bytes as a short summary (byte length and a truncated SHA-256) rather than raw binary.

• shell (bool, default: False ) –

  When True, execute cmd via the system shell (subprocess.run(..., shell=True)). Enables shell features such as pipes, redirections, and glob expansion. The flag is also recorded for display purposes.

• check (bool, default: True ) –

  When True, a subprocess.CalledProcessError is raised on a non-zero exit code. The invocation is recorded before this check, so a failing subprocess is still archived. In case of such a failure, the subprocess’s standard output and error are printed to the caller’s standard output and error stream.

• text (bool | None, default: None ) –

  Whether to run in text or binary mode. By default, the mode follows the type of stdin: text for str, binary for bytes. When no stdin is provided, the default is text mode. In binary mode, the captured standard output and error are recorded as a short summary (byte length and a truncated SHA-256) rather than verbatim, just like binary stdin.

Returns:

• completed –

  The subprocess.CompletedProcess returned by subprocess.run.

Raises:

• CalledProcessError –

  When check is True and the subprocess exits with a non-zero return code.

• TypeError –

  When stdin is not str, bytes, or None, or when stdin is inconsistent with the text flag.

Source code in stepup/core/extapi.py

def run_subprocess(
    cmd: str,
    *,
    workdir: StrPath = ".",
    stdin: str | bytes | None = None,
    shell: bool = False,
    check: bool = True,
    text: bool | None = None,
) -> subprocess.CompletedProcess:
    """Run a subprocess and record it for archival purposes.

    This is the convenience wrapper for the case where an extension step wraps an executable.
    The invocation, its return code, and its captured stdout/stderr
    are recorded via `record_subprocess` (subject to the `STEPUP_MAX_OUTPUT_SIZE` cap).

    Parameters
    ----------
    cmd
        The command line, as a single shell-quoted string.
        When `shell=False` (the default),
        `cmd` is split with `shlex.split` and executed directly (no shell),
        so shell features (pipes, redirections, ...) are not available.
        When `shell=True`, `cmd` is passed as-is to the system shell, which enables shell features.
        As an exception, leading `VAR=value` assignments are extracted and applied
        to the subprocess environment, even when `shell=False`.
        In either case, the caller is responsible for proper quoting.
    workdir
        The working directory of the subprocess as a path or string,
        relative to the step's own working directory.
        It is passed to `subprocess.run` as `cwd`.
    stdin
        Standard input fed to the subprocess, or `None`.
        A `str` is passed to `subprocess.run` as-is and implies `text=True`.
        `bytes` are passed as-is as well and imply `text=False`.
        Inconsistent combinations (e.g. `stdin` is `bytes` but `text=True`) raise a `TypeError`.
        The value is forwarded to `record_subprocess`, which stores `bytes` as a short summary
        (byte length and a truncated SHA-256) rather than raw binary.
    shell
        When `True`, execute `cmd` via the system shell (`subprocess.run(..., shell=True)`).
        Enables shell features such as pipes, redirections, and glob expansion.
        The flag is also recorded for display purposes.
    check
        When `True`, a `subprocess.CalledProcessError` is raised on a non-zero exit code.
        The invocation is recorded **before** this check, so a failing subprocess is still archived.
        In case of such a failure, the subprocess's standard output and error are printed
        to the caller's standard output and error stream.
    text
        Whether to run in text or binary mode.
        By default, the mode follows the type of `stdin`: text for `str`, binary for `bytes`.
        When no `stdin` is provided, the default is text mode.
        In binary mode, the captured standard output and error are recorded
        as a short summary (byte length and a truncated SHA-256) rather than verbatim,
        just like binary `stdin`.

    Returns
    -------
    completed
        The `subprocess.CompletedProcess` returned by `subprocess.run`.

    Raises
    ------
    subprocess.CalledProcessError
        When `check` is `True` and the subprocess exits with a non-zero return code.
    TypeError
        When `stdin` is not `str`, `bytes`, or `None`,
        or when `stdin` is inconsistent with the `text` flag.
    """
    if shell:
        env_overrides = None
    else:
        env_overrides, cmd = extract_env_overrides(cmd)
    text = _resolve_text_mode(stdin, text)
    run_env = dict(os.environ)
    if env_overrides is not None:
        run_env.update(env_overrides)
    cp = subprocess.run(
        cmd if shell else shlex.split(cmd),
        cwd=workdir,
        env=run_env,
        shell=shell,
        input=stdin,
        stdin=None if stdin is not None else subprocess.DEVNULL,
        capture_output=True,
        text=text,
        # Ignoring decoding errors is useful to deal with ill-behaved subprocesses like LaTeX.
        # A non-`None` `errors` switches `Popen` to text mode by itself,
        # so it must stay `None` in binary mode.
        errors="ignore" if text else None,
        check=False,  # handled below, so the subprocess can be recorded with its return code
    )
    record_subprocess(
        cmd,
        cp.returncode,
        workdir=workdir,
        env_overrides=env_overrides,
        shell=shell,
        stdin=stdin,
        stdout=cp.stdout,
        stderr=cp.stderr,
    )
    if check and cp.returncode != 0:
        if cp.stdout:
            sys.stdout.write(cp.stdout if text else cp.stdout.decode())
        if cp.stderr:
            sys.stderr.write(cp.stderr if text else cp.stderr.decode())
        raise subprocess.CalledProcessError(cp.returncode, cmd, cp.stdout, cp.stderr)
    return cp

stepup.core.extapi.record_subprocess(cmd, returncode, *, workdir='.', env_overrides=None, shell=False, stdin=None, stdout=None, stderr=None)

Record a subprocess invocation (already run by the caller) for archival purposes.

This is the low-level escape hatch for wrappers that run the subprocess themselves (e.g. for streaming output, Popen-style pipe interaction, shell features, or conditional invocations).

Most wrappers should use run_subprocess instead.

The recorded metadata is meant to be informative for archival and debugging, not authoritative. Outside a running step (when STEPUP_JOB_I is unset and the RPC client is the dummy one), this function is a no-op.

Parameters:

• cmd (str) –

  The command line, as a single shell-quoted string. The caller is responsible for quoting: build it from parts with shlex.join(parts) when arguments may contain spaces or special characters. The string is stored and displayed verbatim.

• returncode (int) –

  The exit code of the subprocess.

• workdir (StrPath, default: '.' ) –

  The working directory of the subprocess as a path or string, relative to the step’s own working directory. It is translated to be relative to STEPUP_ROOT for storage.

• env_overrides (dict[str, str] | None, default: None ) –

  The environment overlay that the caller applied on top of the inherited environment (only the variables it explicitly set), or None. Only this overlay is stored, not the full resolved environment.

• shell (bool, default: False ) –

  Whether cmd was executed via a shell (i.e. subprocess.run(..., shell=True)). This is stored and used when formatting the invocation for display.

• stdin (str | bytes | None, default: None ) –

  The standard input/output/error of the subprocess, or None when not captured. A str is stored verbatim, subject to the STEPUP_MAX_OUTPUT_SIZE cap. bytes (e.g. a pickle blob) are not stored raw: they are recorded as a short summary (byte length and a truncated SHA-256), since the archival record is TEXT and informative rather than authoritative.

• stdout (str | bytes | None, default: None ) –

  The standard input/output/error of the subprocess, or None when not captured. A str is stored verbatim, subject to the STEPUP_MAX_OUTPUT_SIZE cap. bytes (e.g. a pickle blob) are not stored raw: they are recorded as a short summary (byte length and a truncated SHA-256), since the archival record is TEXT and informative rather than authoritative.

• stderr (str | bytes | None, default: None ) –

  The standard input/output/error of the subprocess, or None when not captured. A str is stored verbatim, subject to the STEPUP_MAX_OUTPUT_SIZE cap. bytes (e.g. a pickle blob) are not stored raw: they are recorded as a short summary (byte length and a truncated SHA-256), since the archival record is TEXT and informative rather than authoritative.

Source code in stepup/core/extapi.py

def record_subprocess(
    cmd: str,
    returncode: int,
    *,
    workdir: StrPath = ".",
    env_overrides: dict[str, str] | None = None,
    shell: bool = False,
    stdin: str | bytes | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
) -> None:
    """Record a subprocess invocation (already run by the caller) for archival purposes.

    This is the low-level escape hatch for wrappers that run the subprocess themselves
    (e.g. for streaming output, `Popen`-style pipe interaction, shell features,
    or conditional invocations).

    Most wrappers should use `run_subprocess` instead.

    The recorded metadata is meant to be informative for archival and debugging, not authoritative.
    Outside a running step (when `STEPUP_JOB_I` is unset and the RPC client is the dummy one),
    this function is a no-op.

    Parameters
    ----------
    cmd
        The command line, as a single shell-quoted string.
        The caller is responsible for quoting: build it from parts with `shlex.join(parts)`
        when arguments may contain spaces or special characters.
        The string is stored and displayed verbatim.
    returncode
        The exit code of the subprocess.
    workdir
        The working directory of the subprocess as a path or string,
        relative to the step's own working directory.
        It is translated to be relative to `STEPUP_ROOT` for storage.
    env_overrides
        The environment **overlay** that the caller applied on top of the inherited environment
        (only the variables it explicitly set), or `None`.
        Only this overlay is stored, not the full resolved environment.
    shell
        Whether `cmd` was executed via a shell (i.e. `subprocess.run(..., shell=True)`).
        This is stored and used when formatting the invocation for display.
    stdin, stdout, stderr
        The standard input/output/error of the subprocess, or `None` when not captured.
        A `str` is stored verbatim, subject to the `STEPUP_MAX_OUTPUT_SIZE` cap.
        `bytes` (e.g. a pickle blob) are not stored raw:
        they are recorded as a short summary (byte length and a truncated SHA-256),
        since the archival record is `TEXT` and informative rather than authoritative.
    """
    # The streams are prepared before the early return below,
    # so that a stream of an unsupported type is rejected whether or not a director is listening.
    stdin_text = _stream_for_record(stdin)
    stdout_text = _stream_for_record(stdout)
    stderr_text = _stream_for_record(stderr)
    job_i = get_job_i()
    if job_i < 0:
        return
    get_rpc_client().call.record_subprocess(
        job_i=job_i,
        cmd=cmd,
        returncode=returncode,
        workdir=translate(workdir),
        env_overrides=env_overrides,
        shell=shell,
        stdin=stdin_text,
        stdout=stdout_text,
        stderr=stderr_text,
    )

stepup.core.extapi.filter_dependencies(paths)

Select the paths retained by ${STEPUP_PATH_FILTER}.

A filter item matches as a plain string prefix of the absolute path, not as a sequence of path components, so the default item -venv also ignores paths under venv2. Because a prefix is anchored at ${STEPUP_ROOT}, the default DEFAULT_PATH_FILTER only covers top-level directories.

Parameters:

• paths (Iterable[StrPath]) –

  An iterable of paths or strings to filter. Relative paths are assumed to be relative to the current working directory.

Returns:

• filtered_paths –

  A collection of paths retained by the filter, relative to the current working directory.

Raises:

• StepUpError –

  When ${STEPUP_PATH_FILTER} contains an item that does not start with + or -.

Source code in stepup/core/extapi.py

def filter_dependencies(paths: Iterable[StrPath]) -> set[Path]:
    """Select the paths retained by `${STEPUP_PATH_FILTER}`.

    A filter item matches as a plain string prefix of the absolute path,
    not as a sequence of path components,
    so the default item `-venv` also ignores paths under `venv2`.
    Because a prefix is anchored at `${STEPUP_ROOT}`,
    the default `DEFAULT_PATH_FILTER` only covers top-level directories.

    Parameters
    ----------
    paths
        An iterable of paths or strings to filter.
        Relative paths are assumed to be relative to the current working directory.

    Returns
    -------
    filtered_paths
        A collection of paths retained by the filter,
        relative to the current working directory.

    Raises
    ------
    StepUpError
        When `${STEPUP_PATH_FILTER}` contains an item that does not start with `+` or `-`.
    """
    # Parse the ${STEPUP_PATH_FILTER} environment variable.
    # The getenv function from StepUp amends the current step to depend on the variable,
    # so that every step using it is re-executed when the variable changes.
    filter_str = getenv("STEPUP_PATH_FILTER", DEFAULT_PATH_FILTER)
    # The two appended items are the catch-all rules:
    # retain everything under `${STEPUP_ROOT}` and ignore everything else.
    # The latter is why every absolute path matches a rule,
    # which makes the `ConsistencyError` below unreachable through the public API.
    filter_str += ":+.:-/"
    rules = []
    stepup_root = get_stepup_root()
    for filter_item in filter_str.split(":"):
        if filter_item == "":
            continue
        if filter_item.startswith("+"):
            keep = True
        elif filter_item.startswith("-"):
            keep = False
        else:
            raise StepUpError(f"Invalid filter item: {filter_item}")
        prefix = Path(filter_item[1:])
        if not prefix.isabs():
            prefix = (stepup_root / prefix).realpath()
        rules.append((prefix, keep))

    # Filter paths according to the rules.
    result = set()
    realpwd = Path.cwd().realpath()
    for path in paths:
        abspath = Path(path).realpath()
        for prefix, keep in rules:
            if abspath.startswith(prefix):
                if keep:
                    result.add(abspath.relpath(realpwd))
                break
        else:
            raise ConsistencyError(f"No matching rule found for path: {path}")
    return result

stepup.core.extapi.get_local_import_paths(script_path=None)

Get all local files from sys.modules.

Parameters:

• script_path (StrPath | None, default: None ) –

  The path of the script that is currently running, or None if unknown. It is excluded from the result, because it is an input of the step by construction.

Returns:

• local_paths –

  A sorted list of paths to local files that are currently imported in sys.modules.

Notes

Files are only included if they match the ${STEPUP_PATH_FILTER} environment variable. Modules without an existing file (built-in, frozen or dynamically created ones) are ignored.

Source code in stepup/core/extapi.py

def get_local_import_paths(script_path: StrPath | None = None) -> list[Path]:
    """Get all local files from `sys.modules`.

    Parameters
    ----------
    script_path
        The path of the script that is currently running, or `None` if unknown.
        It is excluded from the result,
        because it is an input of the step by construction.

    Returns
    -------
    local_paths
        A sorted list of paths to local files that are currently imported in `sys.modules`.

    Notes
    -----
    Files are only included if they match the `${STEPUP_PATH_FILTER}` environment variable.
    Modules without an existing file (built-in, frozen or dynamically created ones)
    are ignored.
    """
    mod_paths = filter_dependencies(_iter_loaded_module_paths())
    if script_path is not None:
        mod_paths.discard(Path(script_path).normpath())
    return sorted(mod_paths)