dwas¶

Everything explicitly exposed here is part of the dwas public API.

In addition to what is here, dwas also exposed the following public modules:

dwas.predefined

This module contains common predefined steps to help reduce duplication.

Warning

While dwas is not at version 1.0.0, it does not guarantee API stability.

Inventory¶

Classes¶

`Config`	Holds the global configuration for `dwas`.
`Step`	Holds the information about a step in the pipeline.
`StepRunner`	Defines the runner for a step, and provides utilities for the step to run.
`StepWithArtifacts`	Defines a step creating artifacts that can be consumed by dependent steps.
`StepWithCleanup`	Defines a step that needs to do custom cleanup.
`StepWithDependentSetup`	Defines a step that will act in the context of its dependent steps.
`StepWithSetup`	Defines a step that needs some setup that can be cached.

Functions¶

`build_parameters`	Generate a `parametrize()` call based on the provided parameters.
`managed_step`	Register the decorated step, and handle installing its dependencies.
`parametrize`	Parametrize the decorated step with the provided values.
`register_managed_step`	Register the provided step, and handle installing its dependencies.
`register_step`	Register the provided step.
`register_step_group`	Register a step group.
`set_defaults`	Set default values for parameters on the given step.
`step`	Register the decorated step and make it available to the pipeline.

Exceptions¶

`BaseDwasException`	Base exception used for exceptions thrown by DWAS.
`DefaultsAlreadySetException`	Exception raised when `set_defaults()` has already been called on an object.
`MismatchedNumberOfParametersException`	Exception raised when the number of parameters does not match the number of ids.
`ParameterConflictException`	Exception raised when values were passed multiple times for the same parameter.

Classes¶

class dwas.Config(cache_path: str, log_path: str | None, *, verbosity: int, colors: bool | None, n_jobs: int, skip_missing_interpreters: bool, skip_setup: bool, skip_run: bool, fail_fast: bool)[source]¶

Bases: object

Holds the global configuration for dwas.

This contains a lot of the configuration that can be set from the command line and can be access in each step to configure their behavior.

cache_path: Path¶

The path to the root of the cache directory used by dwas.

Note that in most cases, you can use the step-specific cache at StepRunner.cache_path and expose data via StepWithArtifacts.gather_artifacts().

colors: bool¶

Whether to use colored output or not for the output.

Determining whether color output is available is not trivial and many programs do it differently.

Here is how dwas does it:

The cli supports –color|–no-color to force the value
Then, it will look for PY_COLORS and enable colors if this is "1", and disable if it is "0". Any other option will abort the program.
Then, it will look if NO_COLORS is set. If so, it will disable colors.
Then, it will look if FORCE_COLOR is set. If so, it will enable colors.
Then, it will detect if this is running in various CIs (currently github actions is supported.) and enable colors if they support it.
Finally, it will look if this is attached to a tty and enable colors if so.

environ: dict[str, str]¶

The environment to use when running commands.

This environment is on purpose minimal, and will only let pass values like

proxies: http_proxy, https_proxy, no_proxy

ca certificates variables: URL_CA_BUNDLE, REQUEST_CA_BUNDLE, SSL_CERT_FILE

language: LANG, LANGUAGE

pip: PIP_INDEX_URL, PIP_EXTRA_INDEX_URL

python: PYTHONHASHSEED

system: PATH, LD_LIBRARY_PATH, TMPDIR

uv: UV_DEFAULT_INDEX, UV_INDEX

If will also forcefully set PY_COLORS and NO_COLOR based on the configuration. See Config.colors.

If PYTHONHASHSEED is not passed when calling dwas, this will set it to a random value and log it to allow repeating the current run.

fail_fast: bool¶: Whether to stop enqueuing more jobs after the first failure or not.

log_path: Path¶: The path to the directory in which logs are stored.

n_jobs: int¶

The number of jobs to run in parallel.

0 will use the number of cpus on the machine as given by multiprocessing.cpu_count().

skip_missing_interpreters: bool¶: Whether to skip when an interpreter is not found, or fail.

skip_run: bool¶

Whether to skip the run part of each step.

This is the reverse of skip_setup, and only runs the setup part.

skip_setup: bool¶: Whether to skip the setup phase of each step.

venvs_path: Path¶: The path to where the virtual environments are stored.

verbosity: int¶

The verbosity level to use.

0 means an equal number of verbose and quiet flags have been passed positive means more verbose, and thus, negative less.

class dwas.Step(*args, **kwargs)[source]¶

Bases: Protocol

Holds the information about a step in the pipeline.

A step is essentially a function that can be called and that has a __name__ attribute.

Steps can then be parametrized by the help of dwas.parametrize(), and registered using register_step(). at which point they are usable by dwas.

Examples:

A step can either be a simple function:

@step()
def my_step(step: StepRunner) -> None:
    step.run(["echo", "hello!"])

Or a class:

# NOTE: you don't need to explicitely inherit from `Step`
class MyStep:
    def __call__(self, step: StepRunner) -> None:
        step.run(["echo", "hello!"])

register_step(MyStep(), name="my_step")

Note

Use whichever method you prefer. Both are supported. Classes tend to be easier for reusability, but functions work well if you just use them in your project.

We will strive to provide both types in examples.

__call__(*args: Any, **kwargs: Any) → None[source]¶

The method to run when the step is invoked.

Parameters:

It can take any amount of parameters, that can be passed by keyword, so positional only arguments are not supported.

Parameters will then be passed using parametrization, with a few specific parameters being reserved by the system. Namely:

step, which is used to pass the StepRunner.
user_args, which is used to pass arguments that the user would have passed on the cli. This can be useful, to allow users to interact with some cli tool your step might be calling.

For passing other arguments, see dwas.parametrize() and dwas.set_defaults().

class dwas.StepRunner[source]¶

Bases: object

Defines the runner for a step, and provides utilities for the step to run.

This is passed as an argument to every step that executes as step.

It provides various utilities to allow the step to run in an isolated, standardized environment.

property cache_path: Path¶

The path to the cache for the current step.

This can be used to store temporary files or any other artifacts.

This will be cleaned up and emptied before the step runs.

property config: Config¶

The global configuration for the current run.

At this point, you should not be modifying it. However, you can use it to act differently on what you are doing. For example, you might want to use the Config.verbosity to configure the output of some commands you run.

get_artifacts(key: str) → list[Any][source]¶

Get the artifacts exported by previous steps for the given key.

See StepWithArtifacts.gather_artifacts() for how to expose artifacts from a step.

Note

This only returns artifacts exported by the direct dependencies of the current step, and does not go recursively. Unless this depends on a step group, in which case it returns the artifacts of all dependencies in the group.

Parameters:: key – The name of the key for which to get the artifacts
Returns:: A list of artifacts, one per step providing artifacts for the given key.

install(*packages: str, sync: bool = False, no_deps: bool = False, force_reinstall: bool = False) → None[source]¶

Install the provided packages in the current environment.

This is a wrapper around the canonical way of installing packages in the provided environment (e.g. pip), so that users don’t need to handle the details when changing the type of virtual environment (e.g. if you wanted to move to conda.).

Examples:

Here’s a few ways of installing packages depending on what you want.

##
# Without syncing files
##

# Install a single package
step.install(["mypy"])

# Install dependencies from a requirements.txt file
step.install(["--requirements=requirements.txt"])

# Install only the package's dependencies but not the package
step.install(["--requirements=pyproject.toml"])

# Install the current package (See dwas.predefined.Package for a better way)
step.install(["."])

# Install a dependency group
step.install(["--group=dev"])

##
# Syncing files from a lockfile
##

# Install only the package's dependencies but not the package
step.install([], sync=True)

# Install a group of the package
step.install(["--only-group=dev"], sync=True)

# Install the dependencies and a group
step.install(["--group=dev"])

Parameters:

packages – which packages to install
sync – Use uv sync instead of pip install to install the dependencies. This ensures is resolves dependencies according to the lock file, and not installing the latest versions. Note that semantics change a bit, so it is not fully interchangeable with sync=False.
no_deps – ignore the package’s dependencies when installing
force_reinstall – force the re-installation of the package and its dependencies even if already installed

Raises:

KeyboardInterrupt – If the user has tried aborting the program and is waiting for it to finish.

property name: str¶: The name of the current step.

property python: str¶: The name of the current python interpreter.

Note

This is not the absolute path to it, just its name.

Run the provided command in the current environment.

This method makes it’s best to ensure the process’ environment is as isolated as possible. It should be used whenever possible, instead of calling subprocess directly.

It will enforce that the first argument of the command is part of the python virtual environment that is specially created for the current step (in the case when there is isolation).

It will also ensure that the environment in which it is run is clean, and will only get environment entries from Config.environ. To add more values, use env.

Parameters:

command – The command to run, as a list of arguments.
cwd – The working directory in which to run the command.
env – Additional environment variables to pass to the process. Those will be merged on top of the Config.environ values and can override them, but not remove them.
external_command – Set to true if you want to run a command that lives outside the current virtual environment. Otherwise, this will fail the command.
silent_on_success – Whether to silence the command’s output if it succeeds, or show it every time.

Returns:

a subprocess.CompletedProcess with stderr and stdout set to None.

Raises:

KeyboardInterrupt – If the user has tried aborting the program and is waiting for it to finish.

set_env(variable: str, value: str) → None[source]¶

Set a specific environment variable for this runner.

This can be useful for example to set environment variables inside another step when setting up a dependent.

Parameters:

variable – The name of the environment variable to set.
value – The value to set the variable to.

class dwas.StepWithArtifacts(*args, **kwargs)[source]¶

Bases: Step, Protocol

Defines a step creating artifacts that can be consumed by dependent steps.

Sometimes, you want to share artifacts between jobs. For example, you might have some pytest runs that generate coverage reports, and then you want to aggregate them together.

This allows a programmatic interface between steps to access artifacts. See StepRunner.get_artifacts() for how to retrieve those artifacts from another step.

Examples:

If you wanted to have multiple pytest steps, and one that aggregates the coverage, you could do:

Tip

This is what the provided dwas.predefined.pytest() step does.

Using functions

@step()
@parametrize(python=["3.9", "3.10"])
def pytest(step: StepRunner) -> None:
    step.run(
        ["pytest"],
        env={
            "COVERAGE_FILE": str(
                step.cache_path.joinpath(step.python, "coverage")
            ),
        },
    )

def gather_artifacts(step: "StepRunner") -> Dict[str, List[Any]]:
    return step.cache_path.joinpath(step.python, "coverage")

pytest.gather_artifacts = gather_artifacts

Using classes

class Pytest:
    def _get_coverage_file(self, step: StepRunner) -> str:
        return str(step.cache_path / "reports" / "coverage")

    def gather_artifacts(self, step: StepRunner) -> Dict[str, List[Any]]:
        return {"coverage_files": [self._get_coverage_file(step)]}

    def __call__(self, step: StepRunner) -> None:
        step.run(
            ["pytest", *args],
            env={"COVERAGE_FILE": self._get_coverage_file(step)},
        )

register_step(parametrize("python", ["3.9", "3.10"])(Pytest()))

And you could then combine and display the coverage like:

@managed_step(dependencies=["coverage"], requires=["pytest"])
def coverage(self, step: StepRunner) -> None:
    env = {"COVERAGE_FILE": str(step.cache_path / "coverage")}

    coverage_files = step.get_artifacts("coverage_files")
    if not coverage_files:
        raise Exception("No coverage files provided. Can't proceed")

    step.run(["coverage", "combine", "--keep", *coverage_files], env=env)
    step.run(["coverage", "html"], env=env)

Tip

The dwas.predefined.coverage() step does roughly this.

gather_artifacts(step: StepRunner) → dict[str, list[Any]][source]¶

Gather all artifacts exposed by this step.

Parameters:: step – The step handler that was used when running the step.
Returns:: A dictionary of artifact key to a list of arbitrary data. This must return a list, as they are merged with other steps’ artifacts into a single list per artifact key.

class dwas.StepWithCleanup(*args, **kwargs)[source]¶

Bases: Step, Protocol

Defines a step that needs to do custom cleanup.

Sometimes, a step might write outside of the dwas-managed cache, in which case it might be desirable for it to be able to clean that directory.

This allows a step to hook on the dwas --clean invocation to cleanup such files.

Examples:

If you wanted to generated documentation, and wanted to have it easily accessible, you could do:

Tip

This is a simplified example of what dwas.predefined.sphinx() does

using functions

@managed_step(dependencies=["sphinx"], output="./build/docs")
def sphinx(step: StepRunner, output: str) -> None:
    step.run(["sphinx-build", "-b=html", "docs/", output])

def clean(output: str) -> None:
    with suppress(FileNotFoundError):
        shutil.rmtree(output)

sphinx.clean = clean

using classes

class Sphinx(Step):
    def __init__(self) -> None:
        self.__name__ = "sphinx"

    def __call__(self, step: StepRunner, output: str) -> None:
        step.run(["sphinx-build", "-b=html", "docs/", output])

    def clean(output: str) -> None:
        with suppress(FileNotFoundError):
            shutil.rmtree(output)

class dwas.StepWithDependentSetup(*args, **kwargs)[source]¶

Bases: Step, Protocol

Defines a step that will act in the context of its dependent steps.

In addition to having a __call__() method, a Step can act in the context of a dependent step, before this one runs.

This allows another step to, for example, install the project it just built into another virtual environment.

Note

This method is called after setup()

Warning

This method is always called when running, and cannot be skipped like setup() by passing --no-setup.

Examples:

You might want to have a step that builds a wheel of your current package and run your tests against it. This could be done like:

Note

A more complete packaging example is provided as dwas.predefined.package()

Using functions

@managed_step(dependencies=["build"])
def package(step: StepRunner) -> None:
    step.run([step.python, "-m", "build", f"--outdir={step.cache_path}"])

def install(self, original_step: StepRunner, current_step: StepRunner) -> None:
    wheels = list(original_step.cache_path.glob("*.whl"))
    # Assuming this is a universal wheel
    assert len(wheels) == 1

    current_step.install(str(wheels[0]))

package.setup_dependent = install

# This can now be used as a dependency
@step(requires=["package"])
def my_step(step: StepRunner) -> None:
    step.run(["myproject", "--help"])

Using a class

class Package:
    def __call__(step: StepRunner) -> None:
        step.run([step.python, "-m", "build", f"--outdir={step.cache_path}"])

    def setup_dependent(
        self,
        original_step: StepRunner,
        current_step: StepRunner,
    ) -> None:
        wheels = list(original_step.cache_path.glob("*.whl"))
        # Assuming this is a universal wheel
        assert len(wheels) == 1

        current_step.install(str(wheels[0]))

register_managed_step(Package(), name="package", dependencies=["build"])

# This can now be used as a dependency
@step(requires=["package"])
def my_step(step: StepRunner) -> None:
    step.run(["myproject", "--help"])

setup_dependent(original_step: StepRunner, current_step: StepRunner) → None[source]¶

Run some logic into a dependent step.

Parameters:

original_step – The original step handler that was used when the step defining this method was called.
current_step – The current step handler, that contains the context of the step that is going to be executed.

class dwas.StepWithSetup(*args, **kwargs)[source]¶

Bases: Step, Protocol

Defines a step that needs some setup that can be cached.

In addition to having a __call__ method, a Step, can implement a setup() function, that gets called before the method.

This can be useful to separate the actual running of the step, from the necessary preparations.

Warning

the setup is meant to contain work that does not necessarily needs to happen at every run of your step, and can be cached in between.

Tip

When running dwas repeatedly, you can pass a --no-setup flag to avoid running those steps again and thus speedup your run.

Examples:

A setup method can be added on a step declaration:

Note

A more complete pytest example is provided as dwas.predefined.pytest().

Using functions

@step()
def pytest(step: StepRunner) -> None:
    step.run(["pytest"])

def install_dependencies(step: StepRunner) -> None:
    step.install("pytest")

pytest.setup = install_dependencies

Using a class

class Pytest:
    def setup(self, step: StepRunner) -> None:
        step.install("pytest")

    def __call__(self, step: StepRunner) -> None:
        step.run(["pytest"])

register_step(Pytest(), name="pytest")

Tip

This is what register_managed_step() does to install your python dependencies.

setup: Callable[[...], None]¶

The setup method that will be invoked before running the step.

This step should run work that is necessary for the test to be able to run but that does not require running every time, as it can be skipped when running dwas with –no-setup.

Parameters:: Parameters are passed to this function the same way they are passed to __call__()

Functions¶

dwas.build_parameters(**kwargs: Any) → Callable[[T], T][source]¶

Generate a parametrize() call based on the provided parameters.

This is a shortcut to build a single parametrize() call, for all non-None values that are passed in.

It will only pass the arguments as a single entry, so this will only ever generate a single entry.

This is basically a shortcut for:

for key, value in parameters.items():
    if value is not None:
        func = parametrize(key, [value])(func)

Parameters:: kwargs – Any key/value pair to pass as a parametrize argument
Returns:: A function to apply the parameters on the given step.

This is a convenience wrapper calling register_managed_step() on the decorated object.

Parameters:

dependencies – A list of dependencies to install as a setup phase. If None, will expect a dependencies parameter to be passed via parametrize().
dependencies_sync – Use uv sync when dealing with dependencies instead of pip install. See StepRunner.install() for examples and details.
name – The name used to refer to this step
description – An optional description of what the current step does
python – The python version to use in this step
requires – The list of steps that this step depends on
run_by_default – Whether this step should run by default or not
passenv – A list of environment variables to pass through to the step.
setenv – A list of environment variables to set in the context of the step.

dwas.parametrize(arg_names: str, args_values: Sequence[Any], ids: Sequence[str | None] | None = None) → Callable[[T], T][source]¶

dwas.parametrize(arg_names: Sequence[str], args_values: Sequence[Sequence[Any]], ids: Sequence[str | None] | None = None) → Callable[[T], T]

Parametrize the decorated step with the provided values.

Parametrization allows running a specific step with multiple configurations. For example, you might want to run a pytest step against both python3.10 and python3.11. With parametrization you do not need to repeat the step.

It is possible to make multiple calls to parametrize on the same step, as long as the parameter names do not conflict. In which case, the product of both parametrization steps will be generated.

Note

Parameters, once set for a specific argument cannot be overridden. If you want to provide default values, see set_defaults().

Parameters:

arg_names – The name of the argument to parametrize. Or a list of names if multiple values need to be passed.
args_values – A list of values to be used for the given argument. When parametrizing multiple arguments at once, this becomes a list of list of argument values.
ids –
A list of ids for each entry in arg_values.

If not provided, it will be either:
- ””, if only one value was passed for the parametrization
- built based on the string representation of the values, in order.

Returns:

A decorator that can be applied to a step to apply the parametrization.

Raises:

MismatchedNumberOfParametersException – if the number of ids and the number of arg_values do not match.
ParameterConflictException – if values have already been provided for a specific argument (e.g. by another parametrize call)

Examples:

You might want to parametrize a single argument. In which case you can do:

# The step needs to be applied after parametrization
@step()
# This step needs to run for both python 3.10 and python3.11
@parametrize("python", ["3.10", "3.11"])
def print_python_version(step: StepRunner) -> None:
    step.execute([self.python, "--version"])

Or you might want to parametrize multiple arguments at once. In that case you can do:

# The step needs to be applied after parametrization.
# Note that we don't supply the usual 'dependencies' argument here, it
# will be handled by parametrization.
@managed_step()
# This needs to run with:
#   - python 3.10 against django 3.0 and 4.0
#   - python3.11 against django 4.0
@parametrize(
    ["python", "dependencies"],
    [
        ["3.10", ["django==3.0"]],
        ["3.10", ["django==4.0"]],
        ["3.11", ["django==4.0"]],
    ],
)
def test(step: StepRunner) -> None:
    step.run([self.python, "manage.py", "test"])

And finally, you can also combine multiple parametrize calls:

# The step needs to be applied after parametrization again.
# Note that we don't supply the usual 'dependencies' argument here, it
# will be handled by parametrization.
@managed_step()
# This needs to run with:
#   - python3.10 and 3.11
#   - both against django 3.0 and 4.0
@parametrize("python", ["3.10", "3.11"])
@parametrize("dependencies", [["django==3.0"], ["django==4.0"]])
def test(step: StepRunner) -> None:
    step.run([self.python, "manage.py", "test"])

A managed step is a step that depends on python packages being present in the environment, and this is taking care of installing the dependencies for it as a setup phase.

Exceptions¶

exception dwas.BaseDwasException(message: str, exit_code: int = 2)[source]¶

Bases: Exception

Base exception used for exceptions thrown by DWAS.

Parameters:

message – A user-facing message explaining what happened.
exit_code –
the exit code to use for the dwas process if the exception is not caught.
- 1 means a pipeline run failed
- 2 means a user or configuration error

exception dwas.DefaultsAlreadySetException(func: Callable[[...], None])[source]¶

Bases: BaseDwasException

Exception raised when set_defaults() has already been called on an object.

exception dwas.MismatchedNumberOfParametersException(n_args_values: int, n_args_ids: int)[source]¶

Bases: BaseDwasException

Exception raised when the number of parameters does not match the number of ids.

exception dwas.ParameterConflictException(parameter: str, func: Callable[[...], None])[source]¶

Bases: BaseDwasException

Exception raised when values were passed multiple times for the same parameter.