run-isolated - API reference

Top-level reexported definitions

run_isolated

Run commands in isolated environments, e.g. Docker containers.

FEATURES = {'run-isolated': VERSION} module-attribute

The list of features supported by the run-isolated library.

VERSION = '0.1.0' module-attribute

The run-isolated library version, semver-like.

Config dataclass

Runtime configuration for the run-isolated library.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass(frozen=True)
class Config:
    """Runtime configuration for the run-isolated library."""

    log: logging.Logger
    """The logger to send diagnostic, informational, and error messages to."""

log instance-attribute

The logger to send diagnostic, informational, and error messages to.

Run commands within a Docker container

run_isolated.rdocker.Config dataclass

Bases: Config

Runtime configuration for the run-i-docker tool.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass(frozen=True)
class Config(defs.Config):
    """Runtime configuration for the `run-i-docker` tool."""

    uid: int
    """The account user ID to use within the container."""

    gid: int
    """The account group ID to use within the container."""

gid instance-attribute

The account group ID to use within the container.

uid instance-attribute

The account user ID to use within the container.

run_isolated.rdocker.ContainerVolume dataclass

A single directory to be mounted within the container.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass(frozen=True)
class ContainerVolume:
    """A single directory to be mounted within the container."""

    external: pathlib.Path
    """The full path to the directory on the host."""

    internal: pathlib.Path
    """The path within the container that the directory will be mounted on."""

    readonly: bool
    """Mount the directory read-only."""

external instance-attribute

The full path to the directory on the host.

internal instance-attribute

The path within the container that the directory will be mounted on.

readonly instance-attribute

Mount the directory read-only.

run_isolated.rdocker.Container dataclass

A representation of a Docker container.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass(frozen=True)
class Container:
    """A representation of a Docker container."""

    cid: str
    """The ID string of the container."""

    @classmethod
    def uninitialized(cls) -> Self:
        """Return an uninitialized container object, not to be used."""
        return cls(cid=_CONTAINER_UNINIT)

    @classmethod
    @contextlib.contextmanager
    def start_container(
        cls,
        cfg: Config,
        container: str,
        *,
        name: str | None = None,
        volumes: list[ContainerVolume] | None = None,
        workdir: pathlib.Path | None = None,
    ) -> Iterator[Self]:
        """Start a Docker container, stop it when done, record it in a `Config` object."""

        def output_lines(cmd: list[pathlib.Path | str]) -> list[str]:
            """Run a command, decode its output as UTF-8 lines."""
            try:
                return subprocess.check_output(cmd, encoding="UTF-8").splitlines()  # noqa: S603
            except OSError as err:
                raise CommandRunError(cmd, err) from err
            except subprocess.CalledProcessError as err:
                raise CommandFailError(cmd, err) from err
            except ValueError as err:
                raise CommandDecodeOutputError(cmd, err) from err

        def start() -> str:
            """Start the container, validate the `docker start` output."""
            cfg.log.debug(
                "Starting a Docker container using the '%(container)s' image",
                {"container": container},
            )
            vol_cmd: Final = list(
                itertools.chain(
                    *(
                        [
                            "--volume",
                            f"{vol.external}:{vol.internal}:{'ro' if vol.readonly else 'rw'}",
                        ]
                        for vol in (volumes or [])
                    ),
                ),
            )
            workdir_cmd: Final[list[pathlib.Path | str]] = (
                ["--workdir", workdir] if workdir is not None else []
            )
            name_cmd: Final = ["--name", name] if name is not None else []
            lines_start: Final = output_lines(
                [
                    "docker",
                    "run",
                    "--detach",
                    "--init",
                    "--pull",
                    "never",
                    "--rm",
                    "--user",
                    f"{cfg.uid}:{cfg.gid}",
                    *vol_cmd,
                    *workdir_cmd,
                    *name_cmd,
                    "--",
                    container,
                    "sleep",
                    "7200",
                ],
            )
            if len(lines_start) != 1 or not _RE_CID.match(lines_start[0]):
                sys.exit(f"Unexpected output from `docker start`: {lines_start!r}")
            return lines_start[0]

        def stop() -> None:
            """Stop the container, validate the `docker stop` output."""
            cfg.log.debug("Stopping the '%(cid)s' Docker container", {"cid": cid})
            lines_stop: Final = output_lines(["docker", "stop", "--", cid])
            expected: Final = [cid]
            if lines_stop != expected:
                sys.exit(
                    f"Unexpected output from `docker stop`: "
                    f"expected {expected!r}, got {lines_stop!r}",
                )

        cid: Final = start()
        cfg.log.debug("Got a Docker container with ID '%(cid)s'", {"cid": cid})

        try:
            yield cls(cid=cid)
        finally:
            stop()

    def run_command(
        self,
        cmd: list[pathlib.Path | str],
        ugid: str | None = None,
        workdir: pathlib.Path | None = None,
    ) -> None:
        """Run a command in the container, check for errors."""
        ugid_cmd: Final = ["--user", ugid] if ugid is not None else []
        workdir_cmd: Final[list[pathlib.Path | str]] = (
            ["--workdir", workdir] if workdir is not None else []
        )
        try:
            subprocess.check_call(  # noqa: S603
                ["docker", "exec", *ugid_cmd, *workdir_cmd, "--", self.cid, *cmd],  # noqa: S607
            )
        except OSError as err:
            raise CommandRunError(cmd, err) from err
        except subprocess.CalledProcessError as err:
            raise CommandFailError(cmd, err) from err
        except ValueError as err:
            raise CommandDecodeOutputError(cmd, err) from err

    def run_command_output(
        self,
        cmd: list[pathlib.Path | str],
        ugid: str | None = None,
        workdir: pathlib.Path | None = None,
    ) -> str:
        """Run a command in the container, check for errors."""
        ugid_cmd: Final = ["--user", ugid] if ugid is not None else []
        workdir_cmd: Final[list[pathlib.Path | str]] = (
            ["--workdir", workdir] if workdir is not None else []
        )
        cmd_exec: Final = ["docker", "exec", *ugid_cmd, *workdir_cmd, "--", self.cid, *cmd]
        try:
            return subprocess.check_output(cmd_exec, encoding="UTF-8")  # noqa: S603
        except OSError as err:
            raise CommandRunError(cmd_exec, err) from err
        except subprocess.CalledProcessError as err:
            raise CommandFailError(cmd_exec, err) from err
        except ValueError as err:
            raise CommandDecodeOutputError(cmd_exec, err) from err

cid instance-attribute

The ID string of the container.

run_command(cmd, ugid=None, workdir=None)

Run a command in the container, check for errors.

Source code in src/run_isolated/rdocker.py

def run_command(
    self,
    cmd: list[pathlib.Path | str],
    ugid: str | None = None,
    workdir: pathlib.Path | None = None,
) -> None:
    """Run a command in the container, check for errors."""
    ugid_cmd: Final = ["--user", ugid] if ugid is not None else []
    workdir_cmd: Final[list[pathlib.Path | str]] = (
        ["--workdir", workdir] if workdir is not None else []
    )
    try:
        subprocess.check_call(  # noqa: S603
            ["docker", "exec", *ugid_cmd, *workdir_cmd, "--", self.cid, *cmd],  # noqa: S607
        )
    except OSError as err:
        raise CommandRunError(cmd, err) from err
    except subprocess.CalledProcessError as err:
        raise CommandFailError(cmd, err) from err
    except ValueError as err:
        raise CommandDecodeOutputError(cmd, err) from err

run_command_output(cmd, ugid=None, workdir=None)

Run a command in the container, check for errors.

Source code in src/run_isolated/rdocker.py

def run_command_output(
    self,
    cmd: list[pathlib.Path | str],
    ugid: str | None = None,
    workdir: pathlib.Path | None = None,
) -> str:
    """Run a command in the container, check for errors."""
    ugid_cmd: Final = ["--user", ugid] if ugid is not None else []
    workdir_cmd: Final[list[pathlib.Path | str]] = (
        ["--workdir", workdir] if workdir is not None else []
    )
    cmd_exec: Final = ["docker", "exec", *ugid_cmd, *workdir_cmd, "--", self.cid, *cmd]
    try:
        return subprocess.check_output(cmd_exec, encoding="UTF-8")  # noqa: S603
    except OSError as err:
        raise CommandRunError(cmd_exec, err) from err
    except subprocess.CalledProcessError as err:
        raise CommandFailError(cmd_exec, err) from err
    except ValueError as err:
        raise CommandDecodeOutputError(cmd_exec, err) from err

start_container(cfg, container, *, name=None, volumes=None, workdir=None) classmethod

Start a Docker container, stop it when done, record it in a Config object.

Source code in src/run_isolated/rdocker.py

@classmethod
@contextlib.contextmanager
def start_container(
    cls,
    cfg: Config,
    container: str,
    *,
    name: str | None = None,
    volumes: list[ContainerVolume] | None = None,
    workdir: pathlib.Path | None = None,
) -> Iterator[Self]:
    """Start a Docker container, stop it when done, record it in a `Config` object."""

    def output_lines(cmd: list[pathlib.Path | str]) -> list[str]:
        """Run a command, decode its output as UTF-8 lines."""
        try:
            return subprocess.check_output(cmd, encoding="UTF-8").splitlines()  # noqa: S603
        except OSError as err:
            raise CommandRunError(cmd, err) from err
        except subprocess.CalledProcessError as err:
            raise CommandFailError(cmd, err) from err
        except ValueError as err:
            raise CommandDecodeOutputError(cmd, err) from err

    def start() -> str:
        """Start the container, validate the `docker start` output."""
        cfg.log.debug(
            "Starting a Docker container using the '%(container)s' image",
            {"container": container},
        )
        vol_cmd: Final = list(
            itertools.chain(
                *(
                    [
                        "--volume",
                        f"{vol.external}:{vol.internal}:{'ro' if vol.readonly else 'rw'}",
                    ]
                    for vol in (volumes or [])
                ),
            ),
        )
        workdir_cmd: Final[list[pathlib.Path | str]] = (
            ["--workdir", workdir] if workdir is not None else []
        )
        name_cmd: Final = ["--name", name] if name is not None else []
        lines_start: Final = output_lines(
            [
                "docker",
                "run",
                "--detach",
                "--init",
                "--pull",
                "never",
                "--rm",
                "--user",
                f"{cfg.uid}:{cfg.gid}",
                *vol_cmd,
                *workdir_cmd,
                *name_cmd,
                "--",
                container,
                "sleep",
                "7200",
            ],
        )
        if len(lines_start) != 1 or not _RE_CID.match(lines_start[0]):
            sys.exit(f"Unexpected output from `docker start`: {lines_start!r}")
        return lines_start[0]

    def stop() -> None:
        """Stop the container, validate the `docker stop` output."""
        cfg.log.debug("Stopping the '%(cid)s' Docker container", {"cid": cid})
        lines_stop: Final = output_lines(["docker", "stop", "--", cid])
        expected: Final = [cid]
        if lines_stop != expected:
            sys.exit(
                f"Unexpected output from `docker stop`: "
                f"expected {expected!r}, got {lines_stop!r}",
            )

    cid: Final = start()
    cfg.log.debug("Got a Docker container with ID '%(cid)s'", {"cid": cid})

    try:
        yield cls(cid=cid)
    finally:
        stop()

uninitialized() classmethod

Return an uninitialized container object, not to be used.

Source code in src/run_isolated/rdocker.py

@classmethod
def uninitialized(cls) -> Self:
    """Return an uninitialized container object, not to be used."""
    return cls(cid=_CONTAINER_UNINIT)

Docker container errors

run_isolated.rdocker.DockerError dataclass

Bases: Exception

An error that occurred while handling the Docker container.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass
class DockerError(Exception):
    """An error that occurred while handling the Docker container."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        raise NotImplementedError

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/rdocker.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    raise NotImplementedError

run_isolated.rdocker.CommandError dataclass

Bases: DockerError

An error related to running a command.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass
class CommandError(DockerError):
    """An error related to running a command."""

    cmd: list[pathlib.Path | str]
    """The command we tried to run."""

    @property
    def cmdstr(self) -> str:
        """Join the words of the command."""
        return pshlex.join(self.cmd)

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return f"Could not run the `{self.cmdstr}` command: {self!r}"

cmd instance-attribute

The command we tried to run.

cmdstr property

Join the words of the command.

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/rdocker.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return f"Could not run the `{self.cmdstr}` command: {self!r}"

run_isolated.rdocker.CommandRunError dataclass

Bases: CommandError

An error that occurred while trying to run a command.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass
class CommandRunError(CommandError):
    """An error that occurred while trying to run a command."""

    err: OSError
    """The error that occurred."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return f"Could not run the `{self.cmdstr}` command: {self.err}"

err instance-attribute

The error that occurred.

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/rdocker.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return f"Could not run the `{self.cmdstr}` command: {self.err}"

run_isolated.rdocker.CommandFailError dataclass

Bases: CommandError

A command that we tried to run failed.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass
class CommandFailError(CommandError):
    """A command that we tried to run failed."""

    err: subprocess.CalledProcessError
    """The error that occurred."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return (
            f"The `{self.cmdstr}` command exited with code {self.err.returncode}; "
            f"output: {self.err.stdout!r}; error output: {self.err.stderr!r}"
        )

err instance-attribute

The error that occurred.

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/rdocker.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return (
        f"The `{self.cmdstr}` command exited with code {self.err.returncode}; "
        f"output: {self.err.stdout!r}; error output: {self.err.stderr!r}"
    )

run_isolated.rdocker.CommandDecodeOutputError dataclass

Bases: CommandError

A command that we tried to run failed.

Source code in src/run_isolated/rdocker.py

@dataclasses.dataclass
class CommandDecodeOutputError(CommandError):
    """A command that we tried to run failed."""

    err: ValueError
    """The error that occurred."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return f"Could not parse the output of the `{self.cmdstr}` command: {self.err}"

err instance-attribute

The error that occurred.

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/rdocker.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return f"Could not parse the output of the `{self.cmdstr}` command: {self.err}"

Helper utility functions

run_isolated.util.DEFAULT_LOGGER_NAME = 'run_isolated' module-attribute

The default name for the logger created by build_logger.

run_isolated.util.build_logger(*, name=DEFAULT_LOGGER_NAME, quiet=False, verbose=False) cached

Build a logger that outputs to the standard output and error streams.

Messages of level WARNING and higher go to the standard error stream. If quiet is false, messages of level INFO go to the standard error stream. If verbose is true, messages of level DEBUG also go to the standard error stream.

Source code in src/run_isolated/util.py

@functools.lru_cache
def build_logger(
    *,
    name: str = DEFAULT_LOGGER_NAME,
    quiet: bool = False,
    verbose: bool = False,
) -> logging.Logger:
    """Build a logger that outputs to the standard output and error streams.

    Messages of level `WARNING` and higher go to the standard error stream.
    If `quiet` is false, messages of level `INFO` go to the standard error stream.
    If `verbose` is true, messages of level `DEBUG` also go to the standard error stream.
    """
    logger: Final = logging.getLogger(name)
    logger.setLevel(logging.DEBUG if verbose else logging.WARNING if quiet else logging.INFO)
    logger.propagate = False

    diag_handler: Final = logging.StreamHandler(sys.stderr)
    diag_handler.setLevel(logging.DEBUG if verbose else logging.WARNING)
    if not quiet:
        diag_handler.addFilter(lambda rec: rec.levelno != logging.INFO)
    logger.addHandler(diag_handler)

    if not quiet:
        info_handler: Final = logging.StreamHandler(sys.stderr)
        info_handler.setLevel(logging.INFO)
        info_handler.addFilter(lambda rec: rec.levelno == logging.INFO)
        logger.addHandler(info_handler)

    return logger