stepup.core.api

You can expect reasonable stability of the API documented here over the future releases of StepUp. (No hard promises, since StepUp is still very young.) Other parts of StepUp, not documented here, may undergo larger changes and are not intended to be API stable.

Basic API

stepup.core.api.static(*paths)

Declare static paths.

Parameters:

• *paths (StrPath | Iterable[StrPath], default: () ) –

  One or more paths to declare as static, relative to the current working directory. Arguments may also be iterables of strings. Each string must refer to an existing file or directory and can be one of:

  1. A file: declared immediately as a static path.
  2. A directory: registered as a static tree; files within it are lazily declared static the first time they are used as step inputs.

Raises:

• ValueError –

  When a path does not exist, when an environment variable in a path is undefined, or when a path contains an invalid variable identifier.

Notes

Environment variables in paths are substituted immediately, and the variables referenced are added to the calling step’s env_deps list. These substitutions are based on the state of os.environ in the calling script, at the time this function is called, not when the step is executed.

Source code in stepup/core/api.py

def static(*paths: StrPath | Iterable[StrPath]):
    """Declare static paths.

    Parameters
    ----------
    *paths
        One or more paths to declare as static, relative to the current working directory.
        Arguments may also be iterables of strings.
        Each string must refer to an existing file or directory and can be one of:

        1. A file: declared immediately as a static path.
        2. A directory: registered as a static tree; files within it are lazily
           declared static the first time they are used as step inputs.

    Raises
    ------
    ValueError
        When a path does not exist, when an environment variable in a path is
        undefined, or when a path contains an invalid variable identifier.

    Notes
    -----
    Environment variables in `paths` are substituted immediately,
    and the variables referenced are added to the calling step's `env_deps` list.
    These substitutions are based on the state of `os.environ` in the calling script,
    at the time this function is called, not when the step is executed.
    """
    # Turn paths into one big list.
    paths = coerce_paths2(paths)

    # Avoid empty RPC calls.
    if len(paths) > 0:
        # Perform env var substitutions.
        with subs_env_vars() as subs:
            su_paths = [subs(path).normpath() for path in paths]
        # Sanity checks
        su_file_paths, su_dir_paths = _check_inp_paths(su_paths, allow_dirs=True)
        if len(su_file_paths) > 0:
            # Translate paths to make them relative to the working directory of the director.
            tr_file_paths = sorted(translate(su_file_path) for su_file_path in su_file_paths)
            # Declare the missing and then confirm the files.
            to_check = RPC_CLIENT.call.declare_missing(_get_step_i(), tr_file_paths)
            _confirm_static(to_check)
        if len(su_dir_paths) > 0:
            # Translate paths to make them relative to the working directory of the director.
            tr_dir_paths = sorted(translate(su_dir_path) for su_dir_path in su_dir_paths)
            # Declare the missing and then confirm the directories.
            to_check = RPC_CLIENT.call.static_trees(_get_step_i(), tr_dir_paths)
            _confirm_deferred(to_check)

stepup.core.api.glob(*patterns, **subs)

Declare static files through glob patterns and return the matches.

All matched files are declared static with the director, and the returned object can be iterated in the calling script.

Parameters:

• *patterns (StrPath, default: () ) –

  One or more glob patterns relative to the current working directory. Patterns may contain anonymous wildcards (*, **) and named wildcards (${*name}).

• **subs (str, default: {} ) –

  Override the sub-pattern matched by each named wildcard. By default every named wildcard matches *.

Returns:

• ngm –

  An NGlobMulti instance with all matched paths. Iteration yields NGlobMatch objects when named wildcards are present, or Path objects when only anonymous wildcards are used. Use ngm.matches() or ngm.files() to force either mode. Use ngm.single() to assert and return exactly one matched path. Evaluates to True in a boolean context when at least one match exists.

Notes

Multiple patterns are matched jointly: only combinations of files whose named wildcard substitutions are mutually consistent are returned. For independent patterns, separate glob calls are more efficient.

Environment variables in patterns are substituted before matching, and the variables referenced are added to the calling step’s env_deps list. These substitutions are based on the state of os.environ in the calling script, at the time this function is called, not when the step is executed.

Source code in stepup/core/api.py

def glob(*patterns: StrPath, **subs: str) -> NGlobMulti:
    """Declare static files through glob patterns and return the matches.

    All matched files are declared static with the director,
    and the returned object can be iterated in the calling script.

    Parameters
    ----------
    *patterns
        One or more glob patterns relative to the current working directory.
        Patterns may contain anonymous wildcards (`*`, `**`) and named wildcards (`${*name}`).
    **subs
        Override the sub-pattern matched by each named wildcard.
        By default every named wildcard matches `*`.

    Returns
    -------
    ngm
        An `NGlobMulti` instance with all matched paths.
        Iteration yields `NGlobMatch` objects when named wildcards are present,
        or `Path` objects when only anonymous wildcards are used.
        Use `ngm.matches()` or `ngm.files()` to force either mode.
        Use `ngm.single()` to assert and return exactly one matched path.
        Evaluates to `True` in a boolean context when at least one match exists.

    Notes
    -----
    Multiple patterns are matched *jointly*: only combinations of files whose
    named wildcard substitutions are mutually consistent are returned.
    For independent patterns, separate `glob` calls are more efficient.

    Environment variables in `patterns` are substituted before matching,
    and the variables referenced are added to the calling step's `env_deps` list.
    These substitutions are based on the state of `os.environ` in the calling script,
    at the time this function is called, not when the step is executed.
    """
    if len(patterns) == 0:
        raise ValueError("At least one path is required for glob.")
    # Substitute environment variables
    with subs_env_vars() as subs_path:
        su_patterns = [subs_path(pattern).normpath() for pattern in patterns]

    # StepUp needs to know the patterns,
    # so it can identify new files matching the patterns in future runs.
    tr_patterns = [translate(su_pattern) for su_pattern in su_patterns]

    # Collect all matches
    nglob_multi = NGlobMulti.from_patterns(su_patterns, subs)
    nglob_multi.glob()

    # Send static paths
    static_paths = nglob_multi.files()
    if len(static_paths) > 0:
        _check_inp_paths(static_paths)
        tr_static_paths = [translate(static_path) for static_path in static_paths]
        to_check = RPC_CLIENT.call.declare_missing(_get_step_i(), tr_static_paths)
        _confirm_static(to_check)

    # Translate all the nglob matches with matching paths and send to the director.
    tr_all_paths = [
        translate(path)
        for nglob_single in nglob_multi.nglob_singles
        for paths in nglob_single.results.values()
        for path in paths
    ]
    RPC_CLIENT.call.nglob(_get_step_i(), tr_patterns, subs, tr_all_paths)

    # Done
    return nglob_multi

stepup.core.api.step(command, *, inp=(), env=(), out=(), vol=(), workdir='.', need=Need.DEFAULT, resources=None, shell=False, _add_exe=False, _need_relative_exe=False)

Add a step to the build graph.

Parameters:

• command (StrPath) –

  Command to execute (in the given working directory). When shell=False, the command may start with one or more VAR=value assignments, e.g. OMP_NUM_THREADS=4 ./run.py. These are stripped from the command and applied as step-specific environment variable overrides when the step runs. This is the only way to set environment variables for a step with shell=False, as there is no shell to interpret such assignments. Note that these overrides (the variable values for the child process) are distinct from env (the variable names the step is sensitive to): a variable may not appear in both, otherwise a ValueError is raised. With shell=True, assignments are left in the command for the shell to interpret. In that case, putting the same variables in env also invalid, but not detected by StepUp.

• inp (Collection[StrPath] | StrPath, default: () ) –

  File(s) required by the step. Relative paths are assumed to be relative to workdir. Directory inputs are not supported.

• env (Collection[str] | str, default: () ) –

  Environment variable(s) to which the step is sensitive. If they change, or when they are (un)defined, the step digest will change, such that the step cannot be skipped.

• out (Collection[StrPath] | StrPath, default: () ) –

  File(s) created by the step. Relative paths are assumed to be relative to workdir. Directory outputs are not supported.

• vol (Collection[StrPath] | StrPath, default: () ) –

  Volatile file(s) created by the step. Relative paths are assumed to be relative to workdir. Directory outputs are not supported.

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

  The directory where the action must be executed. The path is normalized before further processing. If this is a relative path, it is relative to the work directory of the caller. (The default is the current directory.)

• need (Need, default: DEFAULT ) –

  The level of necessity for the step. Three values are allowed: - Need.OPTIONAL = only execute the step if some of its outputs are (indirectly) needed by a non-optional step. - Need.DEFAULT = execute the step unless the user specifies targets. - Need.PLAN = always execute the step because it is part of the plan.

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

  Named resources required to run this step, e.g. {"gpu": 1}. One may also provide the resources as a string, e.g. "gpu:1,memgb:4". The step will not be scheduled until the required units are available, taking into account the units already held by other running steps. Resources not listed in --resources / STEPUP_RESOURCES are treated as unavailable. The required units must be strictly positive and default to 1 when not given, e.g. "gpu" is equivalent to "gpu:1".

• _add_exe (bool, default: False ) –

  (Internal.) When True, if the first word of the command is a local relative executable (it contains a slash and is not absolute), it is added as an input.

• _need_relative_exe (bool, default: False ) –

  (Internal.) When True, require the first word of the command to be such a local relative executable, raising a ValueError otherwise.

Returns:

• step_info –

  Holds relevant information of the step, useful for defining follow-up steps.

Notes

Environment variables in inp, out, vol, and workdir are substituted immediately, and the variables referenced are added to the calling step’s env_deps list. These substitutions are based on the state of os.environ in the calling script, at the time this function is called, not when the step is executed.

Before sending the step to the director, the variables ${inp}, ${out}, and ${vol} in the command are substituted by the space-separated list of inp, out, and vol, respectively. Relative paths in inp, out, and vol are relative to the working directory of the new step.

Source code in stepup/core/api.py

def step(
    command: StrPath,
    *,
    inp: Collection[StrPath] | StrPath = (),
    env: Collection[str] | str = (),
    out: Collection[StrPath] | StrPath = (),
    vol: Collection[StrPath] | StrPath = (),
    workdir: StrPath = ".",
    need: Need = Need.DEFAULT,
    resources: dict[str, int] | str | None = None,
    shell: bool = False,
    _add_exe: bool = False,
    _need_relative_exe: bool = False,
) -> StepInfo:
    """Add a step to the build graph.

    Parameters
    ----------
    command
        Command to execute (in the given working directory).
        When `shell=False`, the command may start with one or more `VAR=value` assignments,
        e.g. `OMP_NUM_THREADS=4 ./run.py`. These are stripped from the command and applied as
        step-specific environment variable overrides when the step runs.
        This is the only way to set environment variables for a step with `shell=False`,
        as there is no shell to interpret such assignments.
        Note that these overrides (the variable **values** for the child process)
        are distinct from `env` (the variable **names** the step is sensitive to):
        a variable may not appear in both, otherwise a `ValueError` is raised.
        With `shell=True`, assignments are left in the command for the shell to interpret.
        In that case, putting the same variables in `env` also invalid, but not detected by StepUp.
    inp
        File(s) required by the step.
        Relative paths are assumed to be relative to `workdir`.
        Directory inputs are not supported.
    env
        Environment variable(s) to which the step is sensitive.
        If they change, or when they are (un)defined, the step digest will change,
        such that the step cannot be skipped.
    out
        File(s) created by the step.
        Relative paths are assumed to be relative to `workdir`.
        Directory outputs are not supported.
    vol
        Volatile file(s) created by the step.
        Relative paths are assumed to be relative to `workdir`.
        Directory outputs are not supported.
    workdir
        The directory where the action must be executed.
        The path is normalized before further processing.
        If this is a relative path, it is relative to the work directory of the caller.
        (The default is the current directory.)
    need
        The level of necessity for the step.
        Three values are allowed:
        - `Need.OPTIONAL` = only execute the step if some of its outputs are (indirectly) needed
          by a non-optional step.
        - `Need.DEFAULT` = execute the step unless the user specifies targets.
        - `Need.PLAN` = always execute the step because it is part of the plan.
    resources
        Named resources required to run this step, e.g. `{"gpu": 1}`.
        One may also provide the resources as a string, e.g. `"gpu:1,memgb:4"`.
        The step will not be scheduled until the required units are available,
        taking into account the units already held by other running steps.
        Resources not listed in `--resources` / `STEPUP_RESOURCES` are treated as unavailable.
        The required units must be strictly positive and default to 1 when not given,
        e.g. `"gpu"` is equivalent to `"gpu:1"`.
    _add_exe
        (Internal.) When `True`, if the first word of the command is a local relative
        executable (it contains a slash and is not absolute), it is added as an input.
    _need_relative_exe
        (Internal.) When `True`, require the first word of the command to be such a local
        relative executable, raising a `ValueError` otherwise.

    Returns
    -------
    step_info
        Holds relevant information of the step, useful for defining follow-up steps.

    Notes
    -----
    Environment variables in `inp`, `out`, `vol`, and `workdir` are substituted immediately,
    and the variables referenced are added to the calling step's `env_deps` list.
    These substitutions are based on the state of `os.environ` in the calling script,
    at the time this function is called, not when the step is executed.

    Before sending the step to the director, the variables `${inp}`, `${out}`, and `${vol}`
    in the command are substituted by the space-separated list of `inp`, `out`, and
    `vol`, respectively.
    Relative paths in `inp`, `out`, and `vol` are relative to the working directory of the new step.
    """
    # Pre-process the arguments for the Director process.
    command = coerce_str(command)
    # Extract leading `VAR=value` assignments as step-specific environment overrides.
    # Only meaningful for shell=False: with a shell, the shell interprets the assignments itself.
    if shell:
        env_overrides = {}
    else:
        env_overrides, command = _extract_env_overrides(command)
    inp_paths = coerce_paths(inp)
    env_deps = string_to_list(env)
    out_paths = coerce_paths(out)
    vol_paths = coerce_paths(vol)

    # Validate the command
    if len(command.strip()) == 0:
        raise ValueError("The command must not be empty.")

    # Validate the environment overrides against the env dependencies and reserved names.
    if env_overrides is not None:
        overlap = set(env_deps) & set(env_overrides)
        if overlap:
            raise ValueError(
                "Variable(s) cannot be both an env dependency and a env_overrides override: "
                + ", ".join(sorted(overlap))
            )
        reserved = set(env_overrides) & RESERVED_ENV_VARS
        if reserved:
            raise ValueError(
                "Variable(s) set by StepUp cannot be overridden: " + ", ".join(sorted(reserved))
            )

    # Detect a local relative executable as the first word of the command.
    auto_exe = None
    if _add_exe:
        try:
            parts = shlex.split(command)
        except ValueError:
            parts = command.split()
        if len(parts) == 0:
            raise ValueError("The command must not be empty.")
        exe = parts[0]
        if os.sep in exe and not exe.startswith(os.sep):
            auto_exe = exe
        elif _need_relative_exe:
            raise ValueError(
                "The command must be a relative path to a local executable, "
                f"containing at least one slash, e.g. './plan.py'. Got: {command}"
            )

    with subs_env_vars() as subs:
        su_inp_paths = [subs(inp_path).normpath() for inp_path in inp_paths]
        su_out_paths = [subs(out_path).normpath() for out_path in out_paths]
        su_vol_paths = [subs(vol_path).normpath() for vol_path in vol_paths]
        su_workdir = subs(workdir).normpath()
        su_exe = None if auto_exe is None else subs(auto_exe).normpath()
    # Substitute paths that are translated back to the current directory.
    # ${inp} expands to the user-specified inputs only, so an auto-added executable does not
    # appear in the expansion (e.g. avoiding a duplicate in `./script.sh ${inp}`).
    command = CaseSensitiveTemplate(command).safe_substitute(
        inp=shlex.join(su_inp_paths),
        out=shlex.join(su_out_paths),
        vol=shlex.join(su_vol_paths),
    )
    # Add the local executable as an input dependency (after the ${inp} expansion above).
    if su_exe is not None:
        su_inp_paths = [su_exe, *su_inp_paths]
    _check_no_directories(su_inp_paths)
    _check_no_directories(su_out_paths)
    _check_no_directories(su_vol_paths)
    tr_inp_paths = [translate(inp_path, su_workdir) for inp_path in su_inp_paths]
    tr_out_paths = [translate(out_path, su_workdir) for out_path in su_out_paths]
    tr_vol_paths = [translate(vol_path, su_workdir) for vol_path in su_vol_paths]
    tr_workdir = translate(su_workdir)

    # Interpret the resources string, if needed.
    if resources is None:
        resources = {}
    elif isinstance(resources, str):
        resources = parse_resources(resources)
    elif not isinstance(resources, dict):
        raise TypeError("The resources argument must be a dict, a string or None.")
    # At this stage, we do not allow non-positive quantities of resources.
    for resource, quantity in resources.items():
        if not isinstance(quantity, int) or quantity <= 0:
            raise ValueError(
                f"Invalid quantity for resource '{resource}': {quantity}. "
                "Must be a strictly positive integer."
            )

    # Warn when a planning step is registered from a non-planning creator.
    if need == Need.PLAN:
        creator_need_name = os.environ.get("STEPUP_STEP_NEED")
        if creator_need_name is not None and creator_need_name != Need.PLAN.name:
            print(
                f"WARNING: planning step '{command}' is registered from a non-planning step"
                f" (creator need={creator_need_name}). This is likely a workflow authoring error.",
                file=sys.stderr,
            )

    # Finally create the step.
    to_check = RPC_CLIENT.call.step(
        _get_step_i(),
        command,
        tr_inp_paths,
        env_deps,
        tr_out_paths,
        tr_vol_paths,
        tr_workdir,
        need.value,
        resources,
        shell,
        env_overrides,
    )

    # Check the existence of files matching static trees.
    _confirm_deferred(to_check)

    # Return a StepInfo instance to facilitate the definition of follow-up steps
    return StepInfo(command, tr_workdir, su_inp_paths, env_deps, su_out_paths, su_vol_paths)

stepup.core.api.amend(*, inp=(), env=(), out=(), vol=())

Declare additional inputs, outputs, and environment dependencies from within a running step.

Parameters:

• inp (Collection[StrPath] | StrPath, default: () ) –

  Files required by the step. Relative paths are relative to the step’s working directory. Directory inputs are not supported.

• env (Collection[str] | str, default: () ) –

  Environment variables to which the step is sensitive. If they change, or when they are (un)defined, the step digest will change, such that the step cannot be skipped.

• out (Collection[StrPath] | StrPath, default: () ) –

  Files created by the step. Relative paths are relative to the step’s working directory. Directory outputs are not supported.

• vol (Collection[StrPath] | StrPath, default: () ) –

  Volatile files created by the step. Relative paths are relative to the step’s working directory. Directory outputs are not supported.

Raises:

• InputNotFoundError –

  When amended inputs are not yet available. Let this exception propagate — do not catch it. The director reschedules the step once the missing inputs become available.

Notes

Environment variables in inp, out, and vol are substituted immediately, and the variables referenced are added to the calling step’s env_deps list. These substitutions are based on the state of os.environ in the calling script, at the time this function is called, not when the step is executed.

Always call amend() before reading input files and before writing output or volatile files.

Repeated calls are safe: items already amended in prior calls are silently skipped.

Source code in stepup/core/api.py

def amend(
    *,
    inp: Collection[StrPath] | StrPath = (),
    env: Collection[str] | str = (),
    out: Collection[StrPath] | StrPath = (),
    vol: Collection[StrPath] | StrPath = (),
):
    """Declare additional inputs, outputs, and environment dependencies from within a running step.

    Parameters
    ----------
    inp
        Files required by the step.
        Relative paths are relative to the step's working directory.
        Directory inputs are not supported.
    env
        Environment variables to which the step is sensitive.
        If they change, or when they are (un)defined, the step digest will change,
        such that the step cannot be skipped.
    out
        Files created by the step.
        Relative paths are relative to the step's working directory.
        Directory outputs are not supported.
    vol
        Volatile files created by the step.
        Relative paths are relative to the step's working directory.
        Directory outputs are not supported.

    Raises
    ------
    InputNotFoundError
        When amended inputs are not yet available.
        Let this exception propagate — do not catch it.
        The director reschedules the step once the missing inputs become available.

    Notes
    -----
    Environment variables in `inp`, `out`, and `vol` are substituted immediately,
    and the variables referenced are added to the calling step's `env_deps` list.
    These substitutions are based on the state of `os.environ` in the calling script,
    at the time this function is called, not when the step is executed.

    Always call `amend()` before reading input files and before writing output or volatile files.

    Repeated calls are safe: items already amended in prior calls are silently skipped.
    """
    # Pre-process the arguments for the Director process.
    inp_paths = coerce_paths(inp)
    env_deps = string_to_list(env)
    out_paths = coerce_paths(out)
    vol_paths = coerce_paths(vol)
    if all(len(collection) == 0 for collection in [inp_paths, env_deps, out_paths, vol_paths]):
        return
    env_deps = set(env_deps)
    with subs_env_vars() as subs:
        su_inp_paths = {subs(inp_path).normpath() for inp_path in inp_paths}
        tr_inp_paths = {translate(inp_path) for inp_path in su_inp_paths}
        tr_out_paths = {translate(subs(out_path)) for out_path in out_paths}
        tr_vol_paths = {translate(subs(vol_path)) for vol_path in vol_paths}
    _check_no_directories(tr_inp_paths)
    _check_no_directories(tr_out_paths)
    _check_no_directories(tr_vol_paths)

    # Filter out previously amended information
    tr_inp_paths.difference_update(AMEND_HISTORY["inp"])
    env_deps.difference_update(AMEND_HISTORY["env"])
    tr_out_paths.difference_update(AMEND_HISTORY["out"])
    tr_vol_paths.difference_update(AMEND_HISTORY["vol"])

    if (
        len(tr_inp_paths) == 0
        and len(env_deps) == 0
        and len(tr_out_paths) == 0
        and len(tr_vol_paths) == 0
    ):
        return

    # Finally, amend for real.
    step_i = _get_step_i()
    amend_result = RPC_CLIENT.call.amend(
        step_i,
        tr_inp_paths,
        sorted(env_deps),
        tr_out_paths,
        tr_vol_paths,
    )
    if amend_result is not None:
        keep_going, to_check = amend_result
        if keep_going is False:
            raise InputNotFoundError("Amended inputs are not available yet.")
        _confirm_deferred(to_check, step_i)

    # Double check that all inputs are indeed present.
    _check_inp_paths(su_inp_paths)

    # Update the amendment history
    AMEND_HISTORY["inp"].update(tr_inp_paths)
    AMEND_HISTORY["env"].update(env_deps)
    AMEND_HISTORY["out"].update(tr_out_paths)
    AMEND_HISTORY["vol"].update(tr_vol_paths)

stepup.core.api.getinfo()

Get the information of the current step.

Returns:

• step_info –

  Holds relevant information of the current step, useful for defining follow-up steps. For consistency with other functions in this module, the inp, out and vol paths are relative to the working directory of the step.

Source code in stepup/core/api.py

def getinfo() -> StepInfo:
    """Get the information of the current step.

    Returns
    -------
    step_info
        Holds relevant information of the current step, useful for defining follow-up steps.
        For consistency with other functions in this module, the `inp`, `out` and `vol`
        paths are relative to the working directory of the step.
    """
    step_info = RPC_CLIENT.call.getinfo(_get_step_i())
    # Update paths to make them relative to the working directory of the step.
    step_info.inp = sorted(translate_back(inp) for inp in step_info.inp)
    step_info.out = sorted(translate_back(out) for out in step_info.out)
    step_info.vol = sorted(translate_back(vol) for vol in step_info.vol)
    return step_info

stepup.core.api.graph(prefix)

Write the workflow graph files in text and dot formats.

Source code in stepup/core/api.py

def graph(prefix: StrPath):
    """Write the workflow graph files in text and dot formats."""
    return RPC_CLIENT.call.graph(coerce_path(prefix))

Composite API

stepup.core.api.run(command, *, inp=(), env=(), out=(), vol=(), workdir='.', optional=False, shell=False, resources=None)

Add a command to the build graph.

Parameters:

• command (StrPath) –

  The command to execute, optionally followed by arguments. The execution method is selected automatically at run time:

  • If shell=True: the command is passed to a shell. Shell features like pipes and redirections are supported.
  • If shell=False and the first word ends in .py: the script is executed via a Python wrapper that auto-detects local imports. Shell features are not available in this mode.
  • If shell=False and the first word is a bare command name (no slashes) that matches a console_scripts entry point in the current Python environment: the entry point is called in-process via the forkserver when available, avoiding subprocess overhead. If the entry point belongs to a different Python environment, a warning is logged and the command falls back to direct subprocess execution.
  • Otherwise: the command is executed directly without a shell. This is faster and safer than the shell mode.

  When the first word contains a / and is not an absolute path (e.g. ./script.py, subdir/tool), it is automatically added as an input dependency. Bare command names like echo or absolute paths like /usr/bin/gcc are not added.

  Python detection uses the .py file extension only, so it works even when the script does not yet exist (e.g. it is an output of another step). shell=True takes precedence and disables Python auto-detection.

• inp (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• env (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• out (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• vol (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• workdir (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• optional (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• resources (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

Returns:

• step_info –

  Holds relevant information of the step, useful for defining follow-up steps.

Source code in stepup/core/api.py

def run(
    command: StrPath,
    *,
    inp: Collection[StrPath] | StrPath = (),
    env: Collection[str] | str = (),
    out: Collection[StrPath] | StrPath = (),
    vol: Collection[StrPath] | StrPath = (),
    workdir: StrPath = ".",
    optional: bool = False,
    shell: bool = False,
    resources: dict[str, int] | str | None = None,
) -> StepInfo:
    """Add a command to the build graph.

    Parameters
    ----------
    command
        The command to execute, optionally followed by arguments.
        The execution method is selected automatically at run time:

        - If `shell=True`: the command is passed to a shell.
          Shell features like pipes and redirections are supported.
        - If `shell=False` and the first word ends in `.py`:
          the script is executed via a Python wrapper
          that auto-detects local imports.
          Shell features are not available in this mode.
        - If `shell=False` and the first word is a bare command name (no slashes) that
          matches a `console_scripts` entry point in the current Python environment:
          the entry point is called in-process via the forkserver when available,
          avoiding subprocess overhead.
          If the entry point belongs to a different Python environment, a warning is
          logged and the command falls back to direct subprocess execution.
        - Otherwise: the command is executed directly without a shell.
          This is faster and safer than the shell mode.

        When the first word contains a `/` and is not an absolute path (e.g. `./script.py`,
        `subdir/tool`), it is automatically added as an input dependency.
        Bare command names like `echo` or absolute paths like `/usr/bin/gcc` are not added.

        Python detection uses the `.py` file extension only,
        so it works even when the script does not yet exist (e.g. it is an output of another step).
        `shell=True` takes precedence and disables Python auto-detection.
    inp, env, out, vol, workdir, optional, resources
        See [`step()`][stepup.core.api.step] for more information.

    Returns
    -------
    step_info
        Holds relevant information of the step, useful for defining follow-up steps.
    """
    return step(
        command,
        inp=inp,
        env=env,
        out=out,
        vol=vol,
        workdir=workdir,
        need=Need.OPTIONAL if optional else Need.DEFAULT,
        resources=resources,
        shell=shell,
        _add_exe=True,
    )

stepup.core.api.plan(command, *, inp=(), env=(), out=(), vol=(), workdir='.', resources=None)

Run a planning script.

The main difference with run() is that the step is flagged as planner internally, which will give it higher priority than non-planner steps. This results in earlier knowledge of the workflow, which improves scheduling efficiency.

Compared to the run() function, this function imposes optional=False and shell=False.

Parameters:

• command (StrPath) –

  The command to execute, optionally followed by arguments. The execution method is selected automatically at run time:

  • If the first word ends in .py: the script is executed via a Python wrapper that auto-detects local imports.
  • Otherwise the command is executed directly without a shell. This scenario is highly unlikely but supported just for completeness.

  Bare command names like echo or absolute paths like /usr/bin/gcc are not allowed. The command must always be a relative path to a local executable script.

• inp (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• env (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• out (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• vol (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• workdir (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• resources (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

Returns:

• step_info –

  Holds relevant information of the step, useful for defining follow-up steps.

Source code in stepup/core/api.py

def plan(
    command: StrPath,
    *,
    inp: Collection[StrPath] | StrPath = (),
    env: Collection[str] | str = (),
    out: Collection[StrPath] | StrPath = (),
    vol: Collection[StrPath] | StrPath = (),
    workdir: StrPath = ".",
    resources: dict[str, int] | str | None = None,
) -> StepInfo:
    """Run a planning script.

    The main difference with [`run()`][stepup.core.api.run] is that the step is flagged
    as planner internally, which will give it higher priority than non-planner steps.
    This results in earlier knowledge of the workflow, which improves scheduling efficiency.

    Compared to the `run()` function, this function imposes `optional=False` and `shell=False`.

    Parameters
    ----------
    command
        The command to execute, optionally followed by arguments.
        The execution method is selected automatically at run time:

        - If the first word ends in `.py`:
          the script is executed via a Python wrapper
          that auto-detects local imports.
        - Otherwise the command is executed directly without a shell.
          This scenario is highly unlikely but supported just for completeness.

        Bare command names like `echo` or absolute paths like `/usr/bin/gcc` are not allowed.
        The command must always be a relative path to a local executable script.
    inp, env, out, vol, workdir, resources
        See [`step()`][stepup.core.api.step] for more information.

    Returns
    -------
    step_info
        Holds relevant information of the step, useful for defining follow-up steps.
    """
    # Note that we do not use `run()` here because we need to set `need=Need.PLAN`.
    return step(
        command,
        inp=inp,
        env=env,
        out=out,
        vol=vol,
        workdir=workdir,
        need=Need.PLAN,
        resources=resources,
        shell=False,
        _add_exe=True,
        _need_relative_exe=True,
    )

stepup.core.api.copy(src, dst, *, optional=False, resources=None)

Add a step that copies a file.

Parameters:

• src (StrPath) –

  This must be a file. Environment variables are substituted.

• dst (StrPath) –

  This can be a file or a directory. Environment variables are substituted. If dst denotes a directory, it must have a trailing slash and src will be copied inside it with its original name. Note that the trailing slash is not supported by pathlib.Path. It is recommended to use a string or path.Path for dst in this case.

• optional (bool, default: False ) –

  See step() for more information.

• resources (bool, default: False ) –

  See step() for more information.

Returns:

• step_info –

  Holds relevant information of the step, useful for defining follow-up steps.

Notes

Environment variables in src and dst are substituted immediately, and the variables referenced are added to the calling step’s env_deps list with amend(). These substitutions are based on the state of os.environ in the calling script, at the time this function is called, not when the copy is actually made.

Source code in stepup/core/api.py

def copy(
    src: StrPath,
    dst: StrPath,
    *,
    optional: bool = False,
    resources: dict[str, int] | str | None = None,
) -> StepInfo:
    """Add a step that copies a file.

    Parameters
    ----------
    src
        This must be a file. Environment variables are substituted.
    dst
        This can be a file or a directory. Environment variables are substituted.
        If `dst` denotes a directory, it must have a trailing slash
        and `src` will be copied inside it with its original name.
        Note that the trailing slash is not supported by `pathlib.Path`.
        It is recommended to use a string or `path.Path` for `dst` in this case.
    optional, resources
        See [`step()`][stepup.core.api.step] for more information.

    Returns
    -------
    step_info
        Holds relevant information of the step, useful for defining follow-up steps.

    Notes
    -----
    Environment variables in `src` and `dst` are substituted immediately,
    and the variables referenced are added to the calling step's `env_deps` list with `amend()`.
    These substitutions are based on the state of `os.environ` in the calling script,
    at the time this function is called, not when the copy is actually made.
    """
    with subs_env_vars() as subs:
        src = subs(src).normpath()
        dst = subs(dst)
    prefix, suffix = get_affixes(dst)
    dst = apply_affixes(dst.normpath(), prefix, suffix)
    dst = make_path_out(src, dst, None)
    return step(
        "cp -p ${inp} ${out}",
        inp=src,
        out=dst,
        need=Need.OPTIONAL if optional else Need.DEFAULT,
        resources=resources,
        shell=False,
    )

stepup.core.api.getenv(name, default=None, *, path=False, back=False, multi=False)

Get an environment variable and amend the current step with the variable name.

Parameters:

• name (str) –

  The name of the environment variable, which is retrieved with os.getenv.

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

  The value to return when the environment variable is unset.

• path (bool, default: False ) –

  Set to True if the variable taken from the environment is assumed to be a path. A Path instance will be returned. Shell variables are substituted (once) in such paths.

• back (bool, default: False ) –

  Set to True to translate the path back to the working directory of the caller. If the path is relative, it is assumed to be relative to the StepUp’s working directory. It will be translated to become relative to the working directory of the caller. This implies path=True.

• multi (bool, default: False ) –

  Set to True if the variable is a list of paths. The paths are split on the colon character and returned as a list of Path instances. This implies path=True.

Returns:

• value –

  The value of the environment variable. If path is set to True, this is a Path instance. If back is set to True, this is a translated Path instance. If multi is set to True, this is a list of Path instances. Otherwise, the result is a string. All path variables are normalized.

Source code in stepup/core/api.py

def getenv(
    name: str,
    default: StrPath | None = None,
    *,
    path: bool = False,
    back: bool = False,
    multi: bool = False,
) -> str | Path | list[Path]:
    """Get an environment variable and amend the current step with the variable name.

    Parameters
    ----------
    name
        The name of the environment variable, which is retrieved with `os.getenv`.
    default
        The value to return when the environment variable is unset.
    path
        Set to True if the variable taken from the environment is assumed to be a path.
        A Path instance will be returned.
        Shell variables are substituted (once) in such paths.
    back
        Set to True to translate the path back to the working directory of the caller.
        If the path is relative, it is assumed to be relative to the StepUp's working directory.
        It will be translated to become relative to the working directory of the caller.
        This implies `path=True`.
    multi
        Set to True if the variable is a list of paths.
        The paths are split on the colon character and returned as a list of `Path` instances.
        This implies `path=True`.

    Returns
    -------
    value
        The value of the environment variable.
        If `path` is set to `True`, this is a `Path` instance.
        If `back` is set to `True`, this is a translated `Path` instance.
        If `multi` is set to `True`, this is a list of `Path` instances.
        Otherwise, the result is a string.
        All path variables are normalized.
    """
    path = path or back or multi
    if default is not None:
        default = coerce_str(default)
    value = os.getenv(name, default)
    # Do not amend environment variables set for the step by the executor.
    # See stepup.core.executor.Executor.run
    if name not in RESERVED_ENV_VARS:
        amend(env=name)

    if multi:
        if value is None:
            return []
        parts = value.split(":")
        value = []
        with subs_env_vars() as subs:
            for item in parts:
                item = item.strip()
                if len(item) > 0:
                    item = subs(item)
                    prefix, suffix = get_affixes(item)
                    item = item.normpath()
                    if back:
                        item = translate_back(item)
                    value.append(apply_affixes(item, prefix, suffix))
    elif path:
        if value is None:
            raise ValueError(f"Undefined shell variable: {name}. Cannot create path.")
        with subs_env_vars() as subs:
            value = subs(value)
        prefix, suffix = get_affixes(value)
        value = value.normpath()
        if back:
            value = translate_back(value)
        value = apply_affixes(value, prefix, suffix)
    return value

stepup.core.api.call(executable_, function_, *, inp=(), env=(), out=(), vol=(), workdir='.', optional=False, planning=False, resources=None, args_file=None, **kwargs)

Register a step that calls a named function in an executable.

Parameters:

• executable_ (StrPath) –

  Path to the script or binary to invoke. Must contain a path separator (e.g. ./script.py or sub/script.py) and must not be an absolute path. Environment variables in the path are substituted.

• function_ (str) –

  Name of the function to invoke (first positional CLI argument).

• inp (Collection[StrPath] | StrPath, default: () ) –

  Files declared as inputs to this step. Normalized to list[str]. Also forwarded to the function as inp.

• env (Collection[str] | str, default: () ) –

  Environment variables tracked by this step.

• out (Collection[StrPath] | StrPath, default: () ) –

  Files declared as outputs of this step. Normalized to list[str]. Also forwarded to the function as out.

• vol (Collection[StrPath] | StrPath, default: () ) –

  Volatile outputs of this step.

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

  Working directory for the step. Defaults to ".".

• optional (bool, default: False ) –

  When True, the step only runs if its outputs are (indirectly) needed by a non-optional step (Need.OPTIONAL). Mutually exclusive with planning.

• planning (bool, default: False ) –

  When True, the step is scheduled as a planner (Need.PLAN). Use this when the called function registers further steps. Mutually exclusive with optional.

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

  Resource constraints for this step.

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

  Full filename for the serialized arguments. When given, arguments are written to this file (format inferred from extension) and passed via --inp=<args_file>; when absent, a JSON string is embedded directly in the command.

• **kwargs –

  Additional keyword arguments forwarded to the function. Must be serializable to JSON by the cattrs JSON converter.

Returns:

• step_info –

  Holds relevant information of the registered step.

Raises:

• ValueError –

  When optional and planning are both True.

• ValueError –

  When executable_ does not contain a path separator or is absolute.

• ValueError –

  When function_ is not a valid Python function name.

• ValueError –

  When the inline JSON string exceeds 128 KiB (use args_file instead).

• ValueError –

  When args_file has an unrecognized extension.

Source code in stepup/core/api.py

def call(
    executable_: StrPath,
    function_: str,
    *,
    inp: Collection[StrPath] | StrPath = (),
    env: Collection[str] | str = (),
    out: Collection[StrPath] | StrPath = (),
    vol: Collection[StrPath] | StrPath = (),
    workdir: StrPath = ".",
    optional: bool = False,
    planning: bool = False,
    resources: dict[str, int] | str | None = None,
    args_file: StrPath | None = None,
    **kwargs,
) -> StepInfo:
    """Register a step that calls a named function in an executable.

    Parameters
    ----------
    executable_
        Path to the script or binary to invoke.
        Must contain a path separator (e.g. `./script.py` or `sub/script.py`)
        and must not be an absolute path.
        Environment variables in the path are substituted.
    function_
        Name of the function to invoke (first positional CLI argument).
    inp
        Files declared as inputs to this step. Normalized to `list[str]`.
        Also forwarded to the function as `inp`.
    env
        Environment variables tracked by this step.
    out
        Files declared as outputs of this step. Normalized to `list[str]`.
        Also forwarded to the function as `out`.
    vol
        Volatile outputs of this step.
    workdir
        Working directory for the step. Defaults to `"."`.
    optional
        When `True`, the step only runs if its outputs are (indirectly) needed
        by a non-optional step (`Need.OPTIONAL`).
        Mutually exclusive with `planning`.
    planning
        When `True`, the step is scheduled as a planner (`Need.PLAN`).
        Use this when the called function registers further steps.
        Mutually exclusive with `optional`.
    resources
        Resource constraints for this step.
    args_file
        Full filename for the serialized arguments.
        When given, arguments are written to this file (format inferred from extension)
        and passed via `--inp=<args_file>`; when absent, a JSON string is embedded
        directly in the command.
    **kwargs
        Additional keyword arguments forwarded to the function.
        Must be serializable to JSON by the `cattrs` JSON converter.

    Returns
    -------
    step_info
        Holds relevant information of the registered step.

    Raises
    ------
    ValueError
        When `optional` and `planning` are both `True`.
    ValueError
        When `executable_` does not contain a path separator or is absolute.
    ValueError
        When `function_` is not a valid Python function name.
    ValueError
        When the inline JSON string exceeds 128 KiB (use `args_file` instead).
    ValueError
        When `args_file` has an unrecognized extension.
    """
    # Validate mutually exclusive flags.
    if optional and planning:
        raise ValueError("optional and planning are mutually exclusive")

    # Perform environment variable substitutions before building the command.
    # This is somewhat redundant with the substitutions performed in `step()`.
    with subs_env_vars() as subs:
        executable_ = subs(executable_)
        inp = [subs(inp_path).normpath() for inp_path in coerce_paths(inp)]
        out = [subs(out_path).normpath() for out_path in coerce_paths(out)]
        workdir = subs(workdir).normpath()
        if args_file is not None:
            args_file = subs(args_file).normpath()
    prefix, suffix = get_affixes(executable_)
    executable_ = apply_affixes(executable_.normpath(), prefix, suffix)

    # Validate executable path format.
    if os.sep not in executable_:
        raise ValueError(
            f"executable_ must contain a path separator (e.g. './script.py'), got: {executable_!r}"
        )

    # Validate the executable is not absolute.
    if os.path.isabs(executable_):
        raise ValueError(f"executable_ must not be an absolute path, got: {executable_!r}")

    # Validate the function name. A valid Python identifier that is not a reserved
    # keyword can never contain shell metacharacters, so it is safe to interpolate
    # unquoted into the command below.
    if not (function_.isidentifier() and not keyword.iskeyword(function_)):
        raise ValueError(f"function_ must be a valid Python function name, got: {function_!r}")

    # Build the forwarded kwargs dict (inp and out are included when not empty).
    forwarded = kwargs.copy()
    if len(inp) > 0:
        forwarded["inp"] = inp
    if len(out) > 0:
        forwarded["out"] = out

    # Build command and step inputs depending on args_file mode.
    if args_file is None:
        unstructured = json_converter.unstructure(forwarded)
        json_str = json.dumps(unstructured)
        if len(json_str.encode()) > 128 * 1024:
            raise ValueError(
                "serialized call arguments exceed 128 KiB; pass args_file= to use a file instead"
            )
        command = f"{shlex.quote(executable_)} {function_} {shlex.quote(json_str)}"
        step_inp = [executable_, *inp]
    else:
        # dumpns(do_amend=True) calls amend(out=args_file) before writing.
        dumpns(args_file, forwarded)
        command = f"{shlex.quote(executable_)} {function_} --inp={shlex.quote(args_file)}"
        step_inp = [executable_, *inp, args_file]

    # Map optional/planning flags to Need enum.
    if optional:
        need = Need.OPTIONAL
    elif planning:
        need = Need.PLAN
    else:
        need = Need.DEFAULT

    # Register and return the step.
    return step(
        command,
        inp=step_inp,
        out=out,
        vol=vol,
        env=env,
        workdir=workdir,
        need=need,
        resources=resources,
    )

stepup.core.api.script(executable, *, step_info=None, inp=(), env=(), out=(), vol=(), workdir='.', optional=False, resources=None)

Run the executable with a single argument plan in a working directory.

Warning

The script interface for calling user Python scripts from plan.py has been deprecated in favor of the new Call interface. You are encouraged to migrate your plan.py files to the new API. See the migration guide for a step-by-step walkthrough.

Parameters:

• executable (StrPath) –

  The path of a local executable that will be called with the argument plan. The file must be executable. The path of the script is assumed to be relative to this directory. Environment variables in the path are substituted.

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

  When given, the steps generated in the plan part of the executable are written to this step_info file. (See stepup.core.stepinfo module for the file format.) This filename is relative to the work directory.

• inp (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• env (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• out (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• vol (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• workdir (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• optional (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

• resources (Collection[StrPath] | StrPath, default: () ) –

  See step() for more information.

Returns:

• step_info –

  Holds relevant information of the step, useful for defining follow-up steps.

Notes

• The arguments inp, env, out and vol are rarely needed for script steps. They only apply to the plan stage of the script, not the run stage.
• The inp argument may be useful when the planning is configured by some input files.
• The optional argument never applies to the plan stage, and is passed on the the run stage.

Source code in stepup/core/api.py

def script(
    executable: StrPath,
    *,
    step_info: StrPath | None = None,
    inp: Collection[StrPath] | StrPath = (),
    env: Collection[str] | str = (),
    out: Collection[StrPath] | StrPath = (),
    vol: Collection[StrPath] | StrPath = (),
    workdir: StrPath = ".",
    optional: bool = False,
    resources: dict[str, int] | str | None = None,
) -> StepInfo:
    """Run the executable with a single argument `plan` in a working directory.

    !!! warning

        The script interface for calling user Python scripts from `plan.py` has been deprecated
        in favor of the new [Call](../getting_started/call.md) interface.
        You are encouraged to migrate your `plan.py` files to the new API.
        See [the migration guide][sc] for a step-by-step walkthrough.

        [sc]: ../migration/from_3x_to_40.md#optional-migration-from-script-to-call

    Parameters
    ----------
    executable
        The path of a local executable that will be called with the argument `plan`.
        The file must be executable.
        The path of the script is assumed to be relative to this directory.
        Environment variables in the path are substituted.
    step_info
        When given, the steps generated in the plan part of the executable are written
        to this `step_info` file. (See [stepup.core.stepinfo][] module for the file format.)
        This filename is relative to the work directory.
    inp, env, out, vol, workdir, optional, resources
        See [`step()`][stepup.core.api.step] for more information.

    Returns
    -------
    step_info
        Holds relevant information of the step, useful for defining follow-up steps.

    Notes
    -----
    - The arguments `inp`, `env`, `out` and `vol` are rarely needed for script steps.
      They only apply to the plan stage of the script, not the run stage.
    - The `inp` argument may be useful when the planning is configured by some input files.
    - The optional argument never applies to the plan stage,
      and is passed on the the run stage.
    """
    # Substitute environment variables in the executable path and normalize it.
    with subs_env_vars() as subs:
        executable = subs(executable)
    prefix, suffix = get_affixes(executable)
    executable = apply_affixes(executable.normpath(), prefix, suffix)

    # Start building the command and the step inputs.
    command = format_command(executable) + " plan"
    out = coerce_paths(out)
    if step_info is not None:
        step_info = coerce_path(step_info)
        command += " --step-info=" + shlex.quote(step_info)
        out.append(step_info)
    if optional:
        command += " --optional"
    inp = coerce_paths(inp)
    inp.append(executable)
    step_kwargs = {
        "inp": inp,
        "env": env,
        "out": out,
        "vol": vol,
        "workdir": workdir,
        "need": Need.PLAN,
        "resources": resources,
    }
    # Note that we do not use `run()` here because we need to set `need=Need.PLAN`.
    return step(command, **step_kwargs)

stepup.core.api.loadns(*paths_variables, dir_out=None, do_amend=True)

Load variable from Python, JSON, TOML or YAML files and put them in a namespace.

Parameters:

• paths_variables (StrPath, default: () ) –

  paths of Python, JSON, TOML or YAML files containing variable definitions. They are loaded in the given order, so later variable definitions may overrule earlier ones. Environment variables in path names are substituted.

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

  This is used to translate paths defined in the variables files (relative to parent of the variable file) to paths relative to dir_out. If not given, the current working directory is used. This is only relevant for variables loaded from Python files.

• do_amend (bool, default: True ) –

  If True, All loaded files are amended as inputs to the current step.

Returns:

• variables –

  A SimpleNamespace instance with the variables, which can be accessed as attributes.

Source code in stepup/core/api.py

def loadns(
    *paths_variables: StrPath, dir_out: StrPath | None = None, do_amend: bool = True
) -> SimpleNamespace:
    """Load variable from Python, JSON, TOML or YAML files and put them in a namespace.

    Parameters
    ----------
    paths_variables
        paths of Python, JSON, TOML or YAML files containing variable definitions.
        They are loaded in the given order, so later variable definitions may overrule earlier ones.
        Environment variables in path names are substituted.
    dir_out
        This is used to translate paths defined in the variables files
        (relative to parent of the variable file)
        to paths relative to `dir_out`.
        If not given, the current working directory is used.
        This is only relevant for variables loaded from Python files.
    do_amend
        If ``True``, All loaded files are amended as inputs to the current step.

    Returns
    -------
    variables
        A SimpleNamespace instance with the variables, which can be accessed as attributes.
    """
    # Process arguments
    dir_out = Path.cwd() if dir_out is None else coerce_path(dir_out)
    with subs_env_vars() as subs:
        paths_variables = [subs(path_var).normpath() for path_var in paths_variables]

    # Build a dictionary of variables
    variables = {}
    for path_var in paths_variables:
        path_var = Path(path_var)
        if path_var.suffix == ".json":
            with open(path_var) as fh:
                variables.update(json.load(fh))
        elif path_var.suffix == ".toml":
            with open(path_var, "rb") as fh:
                variables.update(tomllib.load(fh))
        elif path_var.suffix in (".yaml", ".yml"):
            with open(path_var) as fh:
                variables.update(yaml.safe_load(fh))
        elif path_var.suffix == ".py":
            dir_py = path_var.parent.normpath()
            fn_py = path_var.name
            with contextlib.chdir(dir_py):
                sys.path.insert(0, str(dir_py))
                try:
                    current = run_path(fn_py, run_name="<variables>")
                finally:
                    sys.path.remove(dir_py)
            for name, value in current.items():
                if name.startswith("_"):
                    continue
                if isinstance(value, Path):
                    value = Path(value).relpath(dir_out)
                variables[name] = value
        else:
            raise ValueError(f"unsupported variable file format: {path_var}")
    if do_amend:
        amend(inp=paths_variables)

    # Return as a namespace
    return SimpleNamespace(**variables)

stepup.core.api.dumpns(path, data, *, do_amend=True)

Write variables to a JSON or YAML file.

Parameters:

• path (StrPath) –

  Destination file path. The format is inferred from the extension: .json for JSON, .yaml or .yml for YAML. Environment variables in the path are substituted.

• data (dict | SimpleNamespace) –

  A dict or SimpleNamespace of variables to write. cattrs-supported types (attrs classes, dataclasses) are unstructured automatically.

• do_amend (bool, default: True ) –

  If True, the file is amended as an output of the current step before writing.

Raises:

• ValueError –

  When the file extension is not .json, .yaml, or .yml.

Source code in stepup/core/api.py

def dumpns(path: StrPath, data: dict | SimpleNamespace, *, do_amend: bool = True) -> None:
    """Write variables to a JSON or YAML file.

    Parameters
    ----------
    path
        Destination file path. The format is inferred from the extension:
        `.json` for JSON, `.yaml` or `.yml` for YAML.
        Environment variables in the path are substituted.
    data
        A `dict` or `SimpleNamespace` of variables to write.
        `cattrs`-supported types (attrs classes, dataclasses) are unstructured automatically.
    do_amend
        If `True`, the file is amended as an output of the current step before writing.

    Raises
    ------
    ValueError
        When the file extension is not `.json`, `.yaml`, or `.yml`.
    """
    with subs_env_vars() as subs:
        path = subs(path).normpath()
    if do_amend:
        amend(out=path)
    if isinstance(data, SimpleNamespace):
        data = vars(data)
    path_obj = Path(path)
    if path_obj.suffix == ".json":
        unstructured = json_converter.unstructure(data)
        with open(path_obj, "w") as fh:
            json.dump(unstructured, fh, indent=2)
            fh.write("\n")
    elif path_obj.suffix in (".yaml", ".yml"):
        unstructured = yaml_converter.unstructure(data)
        with open(path_obj, "w") as fh:
            yaml.dump(unstructured, fh)
    else:
        raise ValueError(f"dumpns: unsupported file format: {path_obj.suffix!r}")

stepup.core.api.render_jinja(*args, mode='auto', optional=False, resources=None)

Render the template with Jinja2.

Parameters:

• args (StrPath | dict, default: () ) –

  The first argument is the path to the template file. All the following position arguments can be one of the following two types:

  • Paths to Python, JSON, TOML or YAML files with variable definitions. Variables defined in later files take precedence.
  • A dictionary with additional variables. These will be JSON-serialized and passed on the command-line to the Jinja renderer. Variables in dictionaries take precedence over variables from files. When multiple dictionaries are given, later ones take precedence.

  The very last argument is an output destination (directory or file).

• mode (str, default: 'auto' ) –

  The format of the Jinja placeholders:

  • The default (auto) selects either plain or latex, based on the extension of the output file.
  • The plain format is the default Jinja style with curly brackets: {{ }} etc.
  • The latex style replaces curly brackets by angle brackets: << >> etc.
• optional (bool, default: False ) –

  See step() for more information.

• resources (bool, default: False ) –

  See step() for more information.

Returns:

• step_info –

  Holds relevant information of the step, useful for defining follow-up steps.

Notes

At least some variables must be given, either as a file containing variables or as a dictionary.

Source code in stepup/core/api.py

def render_jinja(
    *args: StrPath | dict,
    mode: str = "auto",
    optional: bool = False,
    resources: dict[str, int] | str | None = None,
) -> StepInfo:
    """Render the template with Jinja2.

    Parameters
    ----------
    args
        The first argument is the path to the template file.
        All the following position arguments can be one of the following two types:

        - Paths to Python, JSON, TOML or YAML files with variable definitions.
          Variables defined in later files take precedence.
        - A dictionary with additional variables.
          These will be JSON-serialized and passed on the command-line to the Jinja renderer.
          Variables in dictionaries take precedence over variables from files.
          When multiple dictionaries are given, later ones take precedence.

        The very last argument is an output destination (directory or file).
    mode
        The format of the Jinja placeholders:

        - The default (auto) selects either `plain` or `latex`,
          based on the extension of the output file.
        - The `plain` format is the default Jinja style with curly brackets: `{{ }}` etc.
        - The `latex` style replaces curly brackets by angle brackets: `<< >>` etc.
    optional, resources
        See [`step()`][stepup.core.api.step] for more information.

    Returns
    -------
    step_info
        Holds relevant information of the step, useful for defining follow-up steps.

    Notes
    -----
    At least some variables must be given, either as a file containing variables or as a dictionary.
    """
    # Parse the positional arguments
    if len(args) < 3:
        raise ValueError(
            "At least three positional arguments must be given: "
            "the template, at least one file or dict with variables, and the destination."
        )
    path_template = args[0]
    if not isinstance(path_template, (str, os.PathLike)):
        raise TypeError("The template argument must be a path.")
    path_template = coerce_path(path_template)
    dest = args[-1]
    if not isinstance(dest, (str, os.PathLike)):
        raise TypeError("The destination argument must be a path.")
    dest = coerce_path(dest)
    variables = {}
    paths_variables = []
    for arg in args[1:-1]:
        if isinstance(arg, dict):
            variables.update(arg)
        elif isinstance(arg, (str, os.PathLike)):
            paths_variables.append(coerce_path(arg))
        else:
            raise TypeError("The variables arguments must be paths or dictionaries.")

    # Parse other arguments.
    if mode not in ["auto", "plain", "latex"]:
        raise ValueError(f"Unsupported mode {mode!r}. Must be one of 'auto', 'plain', 'latex'")
    if len(paths_variables) == 0 and len(variables) == 0:
        raise ValueError("At least one file with variable definitions needed.")
    path_out = make_path_out(path_template, dest, None)

    # Create the command
    args = ["sc-render-jinja", "${inp}", "${out}"]
    if mode != "auto":
        args.append(f"--mode={mode}")
    if len(variables) > 0:
        args.append("--json=" + shlex.quote(json.dumps(variables)))
    return step(
        " ".join(args),
        inp=[path_template, *paths_variables],
        out=path_out,
        need=Need.OPTIONAL if optional else Need.DEFAULT,
        resources=resources,
    )