run-isolated - API reference

Top-level reexported definitions

run_isolated.ContainerVolume dataclass

A single directory to be mounted within the container, if supported.

Source code in src/run_isolated/defs.py

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

    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.Container dataclass

Bases: ABC

A generic implementation of an isolated environment to run commands in.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass(frozen=True)
class Container(abc.ABC):
    """A generic implementation of an isolated environment to run commands in."""

    @classmethod
    @abc.abstractmethod
    def supports_file_copy(cls) -> bool:
        """Report whether this isolated environment supports copying files in and out."""
        raise NotImplementedError

    @classmethod
    @abc.abstractmethod
    def supports_tar_files(cls) -> bool:
        """Report whether this isolated environment supports copying files using `tar`."""
        raise NotImplementedError

    @classmethod
    @abc.abstractmethod
    def supports_volumes(cls) -> bool:
        """Report whether this isolated environment supports mounting volumes in."""
        raise NotImplementedError

    @classmethod
    @abc.abstractmethod
    @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]:
        """Initiaize an isolated environment, stop it when done."""
        raise NotImplementedError

    @abc.abstractmethod
    def run_command(
        self,
        cmd: list[pathlib.Path | str],
        ugid: str | None = None,
        workdir: pathlib.Path | None = None,
    ) -> None:
        """Run a command in the isolated environment, check for errors."""
        raise NotImplementedError

    @abc.abstractmethod
    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 isolated environment, capture its output, check for errors."""
        raise NotImplementedError

    @abc.abstractmethod
    def copy_file_in(self, src: pathlib.Path, dst: pathlib.Path) -> None:
        """Copy a file into the container.

        The ownership and permissions of the file created within the container are
        implementation-dependent.
        """
        raise NotImplementedError

    @abc.abstractmethod
    def copy_file_out(self, src: pathlib.Path, dst: pathlib.Path) -> None:
        """Copy a file out of the container.

        The ownership and permissions of the file created on the host filesystem are
        implementation-dependent.
        """
        raise NotImplementedError

    @abc.abstractmethod
    def tar_files_in(
        self,
        src: pathlib.Path,
        dst_parent: pathlib.Path,
        *,
        ugid: str | None = None,
    ) -> None:
        """Use `tar` to copy a file or directory into the container.

        The ownership and permissions of the files created within the container are
        implementation-dependent.
        """
        raise NotImplementedError

    @abc.abstractmethod
    def tar_files_out(
        self,
        src: pathlib.Path,
        dst_parent: pathlib.Path,
        *,
        ugid: str | None = None,
    ) -> None:
        """Use `tar` to copy a file or directory out of the container.

        The ownership and permissions of the files created on the host filesystem are
        implementation-dependent.
        """
        raise NotImplementedError

copy_file_in(src, dst) abstractmethod

Copy a file into the container.

The ownership and permissions of the file created within the container are implementation-dependent.

Source code in src/run_isolated/defs.py

@abc.abstractmethod
def copy_file_in(self, src: pathlib.Path, dst: pathlib.Path) -> None:
    """Copy a file into the container.

    The ownership and permissions of the file created within the container are
    implementation-dependent.
    """
    raise NotImplementedError

copy_file_out(src, dst) abstractmethod

Copy a file out of the container.

The ownership and permissions of the file created on the host filesystem are implementation-dependent.

Source code in src/run_isolated/defs.py

@abc.abstractmethod
def copy_file_out(self, src: pathlib.Path, dst: pathlib.Path) -> None:
    """Copy a file out of the container.

    The ownership and permissions of the file created on the host filesystem are
    implementation-dependent.
    """
    raise NotImplementedError

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

Run a command in the isolated environment, check for errors.

Source code in src/run_isolated/defs.py

@abc.abstractmethod
def run_command(
    self,
    cmd: list[pathlib.Path | str],
    ugid: str | None = None,
    workdir: pathlib.Path | None = None,
) -> None:
    """Run a command in the isolated environment, check for errors."""
    raise NotImplementedError

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

Run a command in the isolated environment, capture its output, check for errors.

Source code in src/run_isolated/defs.py

@abc.abstractmethod
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 isolated environment, capture its output, check for errors."""
    raise NotImplementedError

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

Initiaize an isolated environment, stop it when done.

Source code in src/run_isolated/defs.py

@classmethod
@abc.abstractmethod
@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]:
    """Initiaize an isolated environment, stop it when done."""
    raise NotImplementedError

supports_file_copy() abstractmethod classmethod

Report whether this isolated environment supports copying files in and out.

Source code in src/run_isolated/defs.py

@classmethod
@abc.abstractmethod
def supports_file_copy(cls) -> bool:
    """Report whether this isolated environment supports copying files in and out."""
    raise NotImplementedError

supports_tar_files() abstractmethod classmethod

Report whether this isolated environment supports copying files using tar.

Source code in src/run_isolated/defs.py

@classmethod
@abc.abstractmethod
def supports_tar_files(cls) -> bool:
    """Report whether this isolated environment supports copying files using `tar`."""
    raise NotImplementedError

supports_volumes() abstractmethod classmethod

Report whether this isolated environment supports mounting volumes in.

Source code in src/run_isolated/defs.py

@classmethod
@abc.abstractmethod
def supports_volumes(cls) -> bool:
    """Report whether this isolated environment supports mounting volumes in."""
    raise NotImplementedError

tar_files_in(src, dst_parent, *, ugid=None) abstractmethod

Use tar to copy a file or directory into the container.

The ownership and permissions of the files created within the container are implementation-dependent.

Source code in src/run_isolated/defs.py

@abc.abstractmethod
def tar_files_in(
    self,
    src: pathlib.Path,
    dst_parent: pathlib.Path,
    *,
    ugid: str | None = None,
) -> None:
    """Use `tar` to copy a file or directory into the container.

    The ownership and permissions of the files created within the container are
    implementation-dependent.
    """
    raise NotImplementedError

tar_files_out(src, dst_parent, *, ugid=None) abstractmethod

Use tar to copy a file or directory out of the container.

The ownership and permissions of the files created on the host filesystem are implementation-dependent.

Source code in src/run_isolated/defs.py

@abc.abstractmethod
def tar_files_out(
    self,
    src: pathlib.Path,
    dst_parent: pathlib.Path,
    *,
    ugid: str | None = None,
) -> None:
    """Use `tar` to copy a file or directory out of the container.

    The ownership and permissions of the files created on the host filesystem are
    implementation-dependent.
    """
    raise NotImplementedError

Common errors

run_isolated.Error dataclass

Bases: Exception

An error that occurred while handling the Docker container.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass
class Error(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/defs.py

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

run_isolated.CommandError dataclass

Bases: Error

An error related to running a command.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass
class CommandError(Error):
    """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/defs.py

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

run_isolated.CommandRunError dataclass

Bases: CommandError

An error that occurred while trying to run a command.

Source code in src/run_isolated/defs.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/defs.py

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

run_isolated.CommandFailError dataclass

Bases: CommandError

A command that we tried to run failed.

Source code in src/run_isolated/defs.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/defs.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.CommandDecodeOutputError dataclass

Bases: CommandError

A command that we tried to run failed.

Source code in src/run_isolated/defs.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/defs.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}"

run_isolated.ConfigNotConfigError dataclass

Bases: Error, TypeError

The supplied cfg parameter was not an instance of the respective Config type.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass
class ConfigNotConfigError(Error, TypeError):
    """The supplied `cfg` parameter was not an instance of the respective `Config` type."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return "The provided `cfg` parameter was not an instance of the expected `Config` class"

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/defs.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return "The provided `cfg` parameter was not an instance of the expected `Config` class"

run_isolated.FileCopyError dataclass

Bases: Error, OSError

Failed to copy a file.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass
class FileCopyError(Error, OSError):
    """Failed to copy a file."""

    src: pathlib.Path
    """The source path we tried to copy."""

    dst: pathlib.Path
    """The destination path we tried to copy to."""

    err: OSError
    """The error we got."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return f"Could not copy {self.src} to {self.dst}: {self.err}"

dst instance-attribute

The destination path we tried to copy to.

err instance-attribute

The error we got.

src instance-attribute

The source path we tried to copy.

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/defs.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return f"Could not copy {self.src} to {self.dst}: {self.err}"

run_isolated.TarFilesError dataclass

Bases: Error

Failed to copy a file.

Source code in src/run_isolated/defs.py

@dataclasses.dataclass
class TarFilesError(Error):
    """Failed to copy a file."""

    src: pathlib.Path
    """The source path we tried to copy."""

    dst_parent: pathlib.Path
    """The destination path we tried to copy to."""

    def __str__(self) -> str:
        """Provide a human-readable error description."""
        return f"Could not copy {self.src} to {self.dst_parent} using tar"

dst_parent instance-attribute

The destination path we tried to copy to.

src instance-attribute

The source path we tried to copy.

__str__()

Provide a human-readable error description.

Source code in src/run_isolated/defs.py

def __str__(self) -> str:
    """Provide a human-readable error description."""
    return f"Could not copy {self.src} to {self.dst_parent} using tar"

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."""

    docker_cmd: str = DEFAULT_DOCKER_CMD
    """The Docker-like command-line tool to use."""

docker_cmd = DEFAULT_DOCKER_CMD class-attribute instance-attribute

The Docker-like command-line tool to use.

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.Container dataclass

Bases: Container

A representation of a Docker container.

Source code in src/run_isolated/rdocker.py

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

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

    cfg: Config
    """The runtime configuration for the container."""

    @classmethod
    def supports_file_copy(cls) -> bool:
        """Report whether this isolated environment supports copying files in and out."""
        return True

    @classmethod
    def supports_tar_files(cls) -> bool:
        """Report whether this isolated environment supports copying files using `tar`."""
        return True

    @classmethod
    def supports_volumes(cls) -> bool:
        """Report whether this isolated environment supports mounting volumes in."""
        return True

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

        def start() -> str:
            """Start the container, validate the `docker start` output."""
            assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
            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 = util.output_lines(
                [
                    cfg.docker_cmd,
                    "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."""
            assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
            cfg.log.debug("Stopping the '%(cid)s' Docker container", {"cid": cid})
            lines_stop: Final = util.output_lines([cfg.docker_cmd, "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}",
                )

        if not isinstance(cfg, Config):
            raise defs.ConfigNotConfigError

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

        try:
            yield cls(cid=cid, cfg=cfg)
        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
                [self.cfg.docker_cmd, "exec", *ugid_cmd, *workdir_cmd, "--", self.cid, *cmd],
            )
        except OSError as err:
            raise defs.CommandRunError(cmd, err) from err
        except subprocess.CalledProcessError as err:
            raise defs.CommandFailError(cmd, err) from err
        except ValueError as err:
            raise defs.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, capture its output, 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 = [
            self.cfg.docker_cmd,
            "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 defs.CommandRunError(cmd_exec, err) from err
        except subprocess.CalledProcessError as err:
            raise defs.CommandFailError(cmd_exec, err) from err
        except ValueError as err:
            raise defs.CommandDecodeOutputError(cmd_exec, err) from err

    def copy_file_in(self, src: pathlib.Path, dst: pathlib.Path) -> None:
        """Copy a file into the container.

        The file created within the container will be owned by the user account that
        the container was started with.
        """
        cmd_cp: Final[list[str | pathlib.Path]] = [
            self.cfg.docker_cmd,
            "cp",
            "--",
            src,
            f"{self.cid}:{dst}",
        ]
        try:
            subprocess.check_call(cmd_cp)  # noqa: S603
        except OSError as err:
            raise defs.CommandRunError(cmd_cp, err) from err
        except subprocess.CalledProcessError as err:
            raise defs.CommandFailError(cmd_cp, err) from err
        except ValueError as err:
            raise defs.CommandDecodeOutputError(cmd_cp, err) from err

    def copy_file_out(self, src: pathlib.Path, dst: pathlib.Path) -> None:
        """Copy a file into the container.

        The file created on the host filesystem will be owned by the user account that
        the current process is running under.
        """
        cmd_cp: Final[list[str | pathlib.Path]] = [
            self.cfg.docker_cmd,
            "cp",
            "--",
            f"{self.cid}:{src}",
            dst,
        ]
        try:
            subprocess.check_call(cmd_cp)  # noqa: S603
        except OSError as err:
            raise defs.CommandRunError(cmd_cp, err) from err
        except subprocess.CalledProcessError as err:
            raise defs.CommandFailError(cmd_cp, err) from err
        except ValueError as err:
            raise defs.CommandDecodeOutputError(cmd_cp, err) from err

    def tar_files_in(
        self,
        src: pathlib.Path,
        dst_parent: pathlib.Path,
        *,
        ugid: str | None = None,
    ) -> None:
        """Use `tar` to copy a file or directory into the container.

        The files created within the container will be owned by the user account that
        the container was started with.
        """
        self.cfg.log.info(
            "Copying %(src)s into the %(cid)s container under %(dst)s",
            {"src": src, "cid": self.cid, "dst": dst_parent},
        )

        ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
        tar_extract: Final = subprocess.Popen(  # noqa: S603
            [
                self.cfg.docker_cmd,
                "exec",
                "-i",
                "-w",
                dst_parent,
                *ugid_cmd,
                "--",
                self.cid,
                "tar",
                "xf",
                "-",
            ],
            stdin=subprocess.PIPE,
            shell=False,
        )
        self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

        tar_create: Final = subprocess.Popen(  # noqa: S603
            ["tar", "cf", "-", src.name],  # noqa: S607
            stdout=tar_extract.stdin,
            shell=False,
            cwd=src.parent,
        )
        self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_create.pid})

        self.cfg.log.debug("- waiting for the create process to complete")
        rc_create: Final = tar_create.wait()
        self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

        self.cfg.log.debug("- waiting for the extract process to complete")
        rc_extract: Final = tar_extract.wait()
        self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

        if rc_create != 0 or rc_extract != 0:
            raise defs.TarFilesError(src, dst_parent)

    def tar_files_out(
        self,
        src: pathlib.Path,
        dst_parent: pathlib.Path,
        ugid: str | None = None,
    ) -> None:
        """Use `tar` to copy a file or directory out of the container.

        The files created on the host filesystem will be owned by the user account that
        the current process is running under.
        """
        self.cfg.log.info(
            "Copying %(src)s out of the %(cid)s container under %(dst)s",
            {"src": src, "cid": self.cid, "dst": dst_parent},
        )

        tar_extract: Final = subprocess.Popen(  # noqa: S603
            ["tar", "xf", "-", src.name],  # noqa: S607
            stdin=subprocess.PIPE,
            shell=False,
            cwd=dst_parent,
        )
        self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

        ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
        tar_create: Final = subprocess.Popen(  # noqa: S603
            [
                self.cfg.docker_cmd,
                "exec",
                "-w",
                src.parent,
                *ugid_cmd,
                "--",
                self.cid,
                "tar",
                "cf",
                "-",
                src.name,
            ],
            stdout=tar_extract.stdin,
            shell=False,
        )
        self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_create.pid})

        self.cfg.log.debug("- waiting for the create process to complete")
        rc_create: Final = tar_create.wait()
        self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

        self.cfg.log.debug("- waiting for the extract process to complete")
        rc_extract: Final = tar_extract.wait()
        self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

        if rc_create != 0 or rc_extract != 0:
            raise defs.TarFilesError(src, dst_parent)

cfg instance-attribute

The runtime configuration for the container.

cid instance-attribute

The ID string of the container.

copy_file_in(src, dst)

Copy a file into the container.

The file created within the container will be owned by the user account that the container was started with.

Source code in src/run_isolated/rdocker.py

def copy_file_in(self, src: pathlib.Path, dst: pathlib.Path) -> None:
    """Copy a file into the container.

    The file created within the container will be owned by the user account that
    the container was started with.
    """
    cmd_cp: Final[list[str | pathlib.Path]] = [
        self.cfg.docker_cmd,
        "cp",
        "--",
        src,
        f"{self.cid}:{dst}",
    ]
    try:
        subprocess.check_call(cmd_cp)  # noqa: S603
    except OSError as err:
        raise defs.CommandRunError(cmd_cp, err) from err
    except subprocess.CalledProcessError as err:
        raise defs.CommandFailError(cmd_cp, err) from err
    except ValueError as err:
        raise defs.CommandDecodeOutputError(cmd_cp, err) from err

copy_file_out(src, dst)

Copy a file into the container.

The file created on the host filesystem will be owned by the user account that the current process is running under.

Source code in src/run_isolated/rdocker.py

def copy_file_out(self, src: pathlib.Path, dst: pathlib.Path) -> None:
    """Copy a file into the container.

    The file created on the host filesystem will be owned by the user account that
    the current process is running under.
    """
    cmd_cp: Final[list[str | pathlib.Path]] = [
        self.cfg.docker_cmd,
        "cp",
        "--",
        f"{self.cid}:{src}",
        dst,
    ]
    try:
        subprocess.check_call(cmd_cp)  # noqa: S603
    except OSError as err:
        raise defs.CommandRunError(cmd_cp, err) from err
    except subprocess.CalledProcessError as err:
        raise defs.CommandFailError(cmd_cp, err) from err
    except ValueError as err:
        raise defs.CommandDecodeOutputError(cmd_cp, err) from err

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
            [self.cfg.docker_cmd, "exec", *ugid_cmd, *workdir_cmd, "--", self.cid, *cmd],
        )
    except OSError as err:
        raise defs.CommandRunError(cmd, err) from err
    except subprocess.CalledProcessError as err:
        raise defs.CommandFailError(cmd, err) from err
    except ValueError as err:
        raise defs.CommandDecodeOutputError(cmd, err) from err

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

Run a command in the container, capture its output, 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, capture its output, 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 = [
        self.cfg.docker_cmd,
        "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 defs.CommandRunError(cmd_exec, err) from err
    except subprocess.CalledProcessError as err:
        raise defs.CommandFailError(cmd_exec, err) from err
    except ValueError as err:
        raise defs.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.

Source code in src/run_isolated/rdocker.py

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

    def start() -> str:
        """Start the container, validate the `docker start` output."""
        assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
        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 = util.output_lines(
            [
                cfg.docker_cmd,
                "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."""
        assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
        cfg.log.debug("Stopping the '%(cid)s' Docker container", {"cid": cid})
        lines_stop: Final = util.output_lines([cfg.docker_cmd, "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}",
            )

    if not isinstance(cfg, Config):
        raise defs.ConfigNotConfigError

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

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

supports_file_copy() classmethod

Report whether this isolated environment supports copying files in and out.

Source code in src/run_isolated/rdocker.py

@classmethod
def supports_file_copy(cls) -> bool:
    """Report whether this isolated environment supports copying files in and out."""
    return True

supports_tar_files() classmethod

Report whether this isolated environment supports copying files using tar.

Source code in src/run_isolated/rdocker.py

@classmethod
def supports_tar_files(cls) -> bool:
    """Report whether this isolated environment supports copying files using `tar`."""
    return True

supports_volumes() classmethod

Report whether this isolated environment supports mounting volumes in.

Source code in src/run_isolated/rdocker.py

@classmethod
def supports_volumes(cls) -> bool:
    """Report whether this isolated environment supports mounting volumes in."""
    return True

tar_files_in(src, dst_parent, *, ugid=None)

Use tar to copy a file or directory into the container.

The files created within the container will be owned by the user account that the container was started with.

Source code in src/run_isolated/rdocker.py

def tar_files_in(
    self,
    src: pathlib.Path,
    dst_parent: pathlib.Path,
    *,
    ugid: str | None = None,
) -> None:
    """Use `tar` to copy a file or directory into the container.

    The files created within the container will be owned by the user account that
    the container was started with.
    """
    self.cfg.log.info(
        "Copying %(src)s into the %(cid)s container under %(dst)s",
        {"src": src, "cid": self.cid, "dst": dst_parent},
    )

    ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
    tar_extract: Final = subprocess.Popen(  # noqa: S603
        [
            self.cfg.docker_cmd,
            "exec",
            "-i",
            "-w",
            dst_parent,
            *ugid_cmd,
            "--",
            self.cid,
            "tar",
            "xf",
            "-",
        ],
        stdin=subprocess.PIPE,
        shell=False,
    )
    self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

    tar_create: Final = subprocess.Popen(  # noqa: S603
        ["tar", "cf", "-", src.name],  # noqa: S607
        stdout=tar_extract.stdin,
        shell=False,
        cwd=src.parent,
    )
    self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_create.pid})

    self.cfg.log.debug("- waiting for the create process to complete")
    rc_create: Final = tar_create.wait()
    self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

    self.cfg.log.debug("- waiting for the extract process to complete")
    rc_extract: Final = tar_extract.wait()
    self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

    if rc_create != 0 or rc_extract != 0:
        raise defs.TarFilesError(src, dst_parent)

tar_files_out(src, dst_parent, ugid=None)

Use tar to copy a file or directory out of the container.

The files created on the host filesystem will be owned by the user account that the current process is running under.

Source code in src/run_isolated/rdocker.py

def tar_files_out(
    self,
    src: pathlib.Path,
    dst_parent: pathlib.Path,
    ugid: str | None = None,
) -> None:
    """Use `tar` to copy a file or directory out of the container.

    The files created on the host filesystem will be owned by the user account that
    the current process is running under.
    """
    self.cfg.log.info(
        "Copying %(src)s out of the %(cid)s container under %(dst)s",
        {"src": src, "cid": self.cid, "dst": dst_parent},
    )

    tar_extract: Final = subprocess.Popen(  # noqa: S603
        ["tar", "xf", "-", src.name],  # noqa: S607
        stdin=subprocess.PIPE,
        shell=False,
        cwd=dst_parent,
    )
    self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

    ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
    tar_create: Final = subprocess.Popen(  # noqa: S603
        [
            self.cfg.docker_cmd,
            "exec",
            "-w",
            src.parent,
            *ugid_cmd,
            "--",
            self.cid,
            "tar",
            "cf",
            "-",
            src.name,
        ],
        stdout=tar_extract.stdin,
        shell=False,
    )
    self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_create.pid})

    self.cfg.log.debug("- waiting for the create process to complete")
    rc_create: Final = tar_create.wait()
    self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

    self.cfg.log.debug("- waiting for the extract process to complete")
    rc_extract: Final = tar_extract.wait()
    self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

    if rc_create != 0 or rc_extract != 0:
        raise defs.TarFilesError(src, dst_parent)

Docker container errors

run_isolated.rdocker.DockerError dataclass

Bases: Error

An error that occurred while handling the Docker container.

Source code in src/run_isolated/rdocker.py

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

Run commands within a schroot environment

run_isolated.rschroot.Config dataclass

Bases: Config

Runtime configuration for the run-i-schroot tool.

Source code in src/run_isolated/rschroot.py

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

    username: str
    """The account username to use within the session."""

    schroot_cmd: str = DEFAULT_SCHROOT_CMD
    """The schroot command-line tool to use."""

schroot_cmd = DEFAULT_SCHROOT_CMD class-attribute instance-attribute

The schroot command-line tool to use.

username instance-attribute

The account username to use within the session.

run_isolated.rschroot.Session dataclass

Bases: Container

A representation of a schroot session.

Source code in src/run_isolated/rschroot.py

@dataclasses.dataclass(frozen=True)
class Session(defs.Container):
    """A representation of a schroot session."""

    sess: str
    """The name of the schroot session."""

    mount_path: pathlib.Path
    """The host directory where the schroot environment's root is mounted."""

    cfg: Config
    """The runtime configuration for the container."""

    @classmethod
    def supports_file_copy(cls) -> bool:
        """Report whether this isolated environment supports copying files in and out."""
        return True

    @classmethod
    def supports_tar_files(cls) -> bool:
        """Report whether this isolated environment supports copying files using `tar`."""
        return True

    @classmethod
    def supports_volumes(cls) -> bool:
        """Report whether this isolated environment supports mounting volumes in."""
        return False

    @classmethod
    @contextlib.contextmanager
    def start_container(
        cls,
        cfg: defs.Config,
        chroot: str,
        *,
        name: str | None = None,
        volumes: list[defs.ContainerVolume] | None = None,
        workdir: pathlib.Path | None = None,
    ) -> Iterator[Self]:
        """Start a schroot session, stop it when done."""

        def start() -> str:
            """Start the session, validate the `schroot -b` output."""
            assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
            cfg.log.debug(
                "Starting a schroot session using the '%(chroot)s' chroot",
                {"chroot": chroot},
            )
            workdir_cmd: Final[list[pathlib.Path | str]] = (
                ["-d", workdir] if workdir is not None else []
            )
            name_cmd: Final = ["-n", name] if name is not None else []
            lines_start: Final = util.output_lines(
                [
                    cfg.schroot_cmd,
                    "-b",
                    "-c",
                    chroot,
                    "-u",
                    cfg.username,
                    *workdir_cmd,
                    *name_cmd,
                ],
            )
            if len(lines_start) != 1 or not _RE_SID.match(lines_start[0]):
                sys.exit(f"Unexpected output from `schroot -b`: {lines_start!r}")
            return lines_start[0]

        def stop() -> None:
            """Stop the session, validate the `schroot -e` output."""
            assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
            cfg.log.debug("Stopping the '%(sess)s' schroot session", {"sess": sess})
            lines_stop: Final = util.output_lines([cfg.schroot_cmd, "-e", "-c", sess])
            expected: Final[list[str]] = []
            if lines_stop != expected:
                sys.exit(
                    f"Unexpected output from `schroot -e`: "
                    f"expected {expected!r}, got {lines_stop!r}",
                )

        if not isinstance(cfg, Config):
            raise defs.ConfigNotConfigError

        if volumes:
            raise NotImplementedError

        sess: Final = start()
        cfg.log.debug("Got a schroot session '%(sess)s'", {"sess": sess})

        try:
            mount_path: Final = cls.get_mount_path(cfg, sess)
            yield cls(sess=sess, mount_path=mount_path, cfg=cfg)
        finally:
            stop()

    @classmethod
    def get_mount_path(cls, cfg: Config, sess: str) -> pathlib.Path:
        """Query schroot for the environment's root directory."""

        def extract_path(line: str) -> pathlib.Path | None:
            """Grab the path out of a line."""
            if not line or not line[0].isspace():
                return None

            match line.split():
                case ["Path", path]:
                    return pathlib.Path(path)

                case _:
                    return None

        cfg.log.info("Looking for the mount path of the %(sess)s session", {"sess": sess})
        lines: Final = util.output_lines([cfg.schroot_cmd, "-i", "-c", f"session:{sess}"])
        paths: Final = [path for path in (extract_path(line) for line in lines) if path is not None]
        cfg.log.info("- got paths %(paths)r", {"paths": paths})
        match paths:
            case [single]:
                return single

            case _:
                raise RuntimeError(repr((lines, paths)))

    def run_command(
        self,
        cmd: list[pathlib.Path | str],
        ugid: str | None = None,
        workdir: pathlib.Path | None = None,
    ) -> None:
        """Run a command in the schroot environment, check for errors."""
        ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
        workdir_cmd: Final = ["-d", str(workdir) if workdir is not None else "/"]
        try:
            subprocess.check_call(  # noqa: S603
                [self.cfg.schroot_cmd, "-r", "-c", self.sess, *ugid_cmd, *workdir_cmd, "--", *cmd],
            )
        except OSError as err:
            raise defs.CommandRunError(cmd, err) from err
        except subprocess.CalledProcessError as err:
            raise defs.CommandFailError(cmd, err) from err
        except ValueError as err:
            raise defs.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 schroot environment, capture its output, check for errors."""
        ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
        workdir_cmd: Final = ["-d", str(workdir) if workdir is not None else "/"]
        try:
            return subprocess.check_output(  # noqa: S603
                [self.cfg.schroot_cmd, "-r", "-c", self.sess, *ugid_cmd, *workdir_cmd, "--", *cmd],
                encoding="UTF-8",
            )
        except OSError as err:
            raise defs.CommandRunError(cmd, err) from err
        except subprocess.CalledProcessError as err:
            raise defs.CommandFailError(cmd, err) from err
        except ValueError as err:
            raise defs.CommandDecodeOutputError(cmd, err) from err

    def copy_file_in(self, src: pathlib.Path, dst: pathlib.Path) -> None:
        """Copy a file into the container.

        The file created within the schroot environment will be owned by the user account that
        the current process is running under.
        """
        dst_full: Final = self.mount_path / dst.relative_to(dst.root)
        try:
            shutil.copy2(src, dst_full)
        except OSError as err:
            raise defs.FileCopyError(src=src, dst=dst_full, err=err) from err

    def copy_file_out(self, src: pathlib.Path, dst: pathlib.Path) -> None:
        """Copy a file into the container.

        The file created on the host filesystem will be owned by the user account that
        the current process is running under.
        """
        src_full: Final = self.mount_path / src.relative_to(src.root)
        try:
            shutil.copy2(src_full, dst)
        except OSError as err:
            raise defs.FileCopyError(src=src_full, dst=dst, err=err) from err

    def tar_files_in(
        self,
        src: pathlib.Path,
        dst_parent: pathlib.Path,
        *,
        ugid: str | None = None,
    ) -> None:
        """Use `tar` to copy a file or directory into the container.

        The files created within the schroot environment will be owned by the user account that
        the current process is running under.
        """
        self.cfg.log.info(
            "Copying %(src)s into the %(sess)s session under %(dst)s",
            {"src": src, "sess": self.sess, "dst": dst_parent},
        )

        ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
        tar_extract: Final = subprocess.Popen(  # noqa: S603
            [
                self.cfg.schroot_cmd,
                "-r",
                "-c",
                self.sess,
                "-d",
                dst_parent,
                *ugid_cmd,
                "--",
                "tar",
                "xf",
                "-",
            ],
            stdin=subprocess.PIPE,
            shell=False,
        )
        self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

        tar_create: Final = subprocess.Popen(  # noqa: S603
            ["tar", "cf", "-", src.name],  # noqa: S607
            stdout=tar_extract.stdin,
            shell=False,
            cwd=src.parent,
        )
        self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_extract.pid})

        self.cfg.log.debug("- waiting for the create process to complete")
        rc_create: Final = tar_create.wait()
        self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

        self.cfg.log.debug("- waiting for the extract process to complete")
        rc_extract: Final = tar_extract.wait()
        self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

        if rc_create != 0 or rc_extract != 0:
            raise defs.TarFilesError(src, dst_parent)

    def tar_files_out(
        self,
        src: pathlib.Path,
        dst_parent: pathlib.Path,
        ugid: str | None = None,
    ) -> None:
        """Use `tar` to copy a file or directory out of the container.

        The files created on the host filesystem will be owned by the user account that
        the current process is running under.
        """
        self.cfg.log.info(
            "Copying %(src)s out of the %(sess)s session under %(dst)s",
            {"src": src, "sess": self.sess, "dst": dst_parent},
        )

        tar_extract: Final = subprocess.Popen(  # noqa: S603
            ["tar", "xf", "-", src.name],  # noqa: S607
            stdin=subprocess.PIPE,
            shell=False,
            cwd=dst_parent,
        )
        self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

        ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
        tar_create: Final = subprocess.Popen(  # noqa: S603
            [
                self.cfg.schroot_cmd,
                "-r",
                "-c",
                self.sess,
                "-d",
                src.parent,
                *ugid_cmd,
                "--",
                "tar",
                "cf",
                "-",
                src.name,
            ],
            stdout=tar_extract.stdin,
            shell=False,
        )
        self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_extract.pid})

        self.cfg.log.debug("- waiting for the create process to complete")
        rc_create: Final = tar_create.wait()
        self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

        self.cfg.log.debug("- waiting for the extract process to complete")
        rc_extract: Final = tar_extract.wait()
        self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

        if rc_create != 0 or rc_extract != 0:
            raise defs.TarFilesError(src, dst_parent)

cfg instance-attribute

The runtime configuration for the container.

mount_path instance-attribute

The host directory where the schroot environment's root is mounted.

sess instance-attribute

The name of the schroot session.

copy_file_in(src, dst)

Copy a file into the container.

The file created within the schroot environment will be owned by the user account that the current process is running under.

Source code in src/run_isolated/rschroot.py

def copy_file_in(self, src: pathlib.Path, dst: pathlib.Path) -> None:
    """Copy a file into the container.

    The file created within the schroot environment will be owned by the user account that
    the current process is running under.
    """
    dst_full: Final = self.mount_path / dst.relative_to(dst.root)
    try:
        shutil.copy2(src, dst_full)
    except OSError as err:
        raise defs.FileCopyError(src=src, dst=dst_full, err=err) from err

copy_file_out(src, dst)

Copy a file into the container.

The file created on the host filesystem will be owned by the user account that the current process is running under.

Source code in src/run_isolated/rschroot.py

def copy_file_out(self, src: pathlib.Path, dst: pathlib.Path) -> None:
    """Copy a file into the container.

    The file created on the host filesystem will be owned by the user account that
    the current process is running under.
    """
    src_full: Final = self.mount_path / src.relative_to(src.root)
    try:
        shutil.copy2(src_full, dst)
    except OSError as err:
        raise defs.FileCopyError(src=src_full, dst=dst, err=err) from err

get_mount_path(cfg, sess) classmethod

Query schroot for the environment's root directory.

Source code in src/run_isolated/rschroot.py

@classmethod
def get_mount_path(cls, cfg: Config, sess: str) -> pathlib.Path:
    """Query schroot for the environment's root directory."""

    def extract_path(line: str) -> pathlib.Path | None:
        """Grab the path out of a line."""
        if not line or not line[0].isspace():
            return None

        match line.split():
            case ["Path", path]:
                return pathlib.Path(path)

            case _:
                return None

    cfg.log.info("Looking for the mount path of the %(sess)s session", {"sess": sess})
    lines: Final = util.output_lines([cfg.schroot_cmd, "-i", "-c", f"session:{sess}"])
    paths: Final = [path for path in (extract_path(line) for line in lines) if path is not None]
    cfg.log.info("- got paths %(paths)r", {"paths": paths})
    match paths:
        case [single]:
            return single

        case _:
            raise RuntimeError(repr((lines, paths)))

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

Run a command in the schroot environment, check for errors.

Source code in src/run_isolated/rschroot.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 schroot environment, check for errors."""
    ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
    workdir_cmd: Final = ["-d", str(workdir) if workdir is not None else "/"]
    try:
        subprocess.check_call(  # noqa: S603
            [self.cfg.schroot_cmd, "-r", "-c", self.sess, *ugid_cmd, *workdir_cmd, "--", *cmd],
        )
    except OSError as err:
        raise defs.CommandRunError(cmd, err) from err
    except subprocess.CalledProcessError as err:
        raise defs.CommandFailError(cmd, err) from err
    except ValueError as err:
        raise defs.CommandDecodeOutputError(cmd, err) from err

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

Run a command in the schroot environment, capture its output, check for errors.

Source code in src/run_isolated/rschroot.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 schroot environment, capture its output, check for errors."""
    ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
    workdir_cmd: Final = ["-d", str(workdir) if workdir is not None else "/"]
    try:
        return subprocess.check_output(  # noqa: S603
            [self.cfg.schroot_cmd, "-r", "-c", self.sess, *ugid_cmd, *workdir_cmd, "--", *cmd],
            encoding="UTF-8",
        )
    except OSError as err:
        raise defs.CommandRunError(cmd, err) from err
    except subprocess.CalledProcessError as err:
        raise defs.CommandFailError(cmd, err) from err
    except ValueError as err:
        raise defs.CommandDecodeOutputError(cmd, err) from err

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

Start a schroot session, stop it when done.

Source code in src/run_isolated/rschroot.py

@classmethod
@contextlib.contextmanager
def start_container(
    cls,
    cfg: defs.Config,
    chroot: str,
    *,
    name: str | None = None,
    volumes: list[defs.ContainerVolume] | None = None,
    workdir: pathlib.Path | None = None,
) -> Iterator[Self]:
    """Start a schroot session, stop it when done."""

    def start() -> str:
        """Start the session, validate the `schroot -b` output."""
        assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
        cfg.log.debug(
            "Starting a schroot session using the '%(chroot)s' chroot",
            {"chroot": chroot},
        )
        workdir_cmd: Final[list[pathlib.Path | str]] = (
            ["-d", workdir] if workdir is not None else []
        )
        name_cmd: Final = ["-n", name] if name is not None else []
        lines_start: Final = util.output_lines(
            [
                cfg.schroot_cmd,
                "-b",
                "-c",
                chroot,
                "-u",
                cfg.username,
                *workdir_cmd,
                *name_cmd,
            ],
        )
        if len(lines_start) != 1 or not _RE_SID.match(lines_start[0]):
            sys.exit(f"Unexpected output from `schroot -b`: {lines_start!r}")
        return lines_start[0]

    def stop() -> None:
        """Stop the session, validate the `schroot -e` output."""
        assert isinstance(cfg, Config)  # noqa: S101  # mypy needs this
        cfg.log.debug("Stopping the '%(sess)s' schroot session", {"sess": sess})
        lines_stop: Final = util.output_lines([cfg.schroot_cmd, "-e", "-c", sess])
        expected: Final[list[str]] = []
        if lines_stop != expected:
            sys.exit(
                f"Unexpected output from `schroot -e`: "
                f"expected {expected!r}, got {lines_stop!r}",
            )

    if not isinstance(cfg, Config):
        raise defs.ConfigNotConfigError

    if volumes:
        raise NotImplementedError

    sess: Final = start()
    cfg.log.debug("Got a schroot session '%(sess)s'", {"sess": sess})

    try:
        mount_path: Final = cls.get_mount_path(cfg, sess)
        yield cls(sess=sess, mount_path=mount_path, cfg=cfg)
    finally:
        stop()

supports_file_copy() classmethod

Report whether this isolated environment supports copying files in and out.

Source code in src/run_isolated/rschroot.py

@classmethod
def supports_file_copy(cls) -> bool:
    """Report whether this isolated environment supports copying files in and out."""
    return True

supports_tar_files() classmethod

Report whether this isolated environment supports copying files using tar.

Source code in src/run_isolated/rschroot.py

@classmethod
def supports_tar_files(cls) -> bool:
    """Report whether this isolated environment supports copying files using `tar`."""
    return True

supports_volumes() classmethod

Report whether this isolated environment supports mounting volumes in.

Source code in src/run_isolated/rschroot.py

@classmethod
def supports_volumes(cls) -> bool:
    """Report whether this isolated environment supports mounting volumes in."""
    return False

tar_files_in(src, dst_parent, *, ugid=None)

Use tar to copy a file or directory into the container.

The files created within the schroot environment will be owned by the user account that the current process is running under.

Source code in src/run_isolated/rschroot.py

def tar_files_in(
    self,
    src: pathlib.Path,
    dst_parent: pathlib.Path,
    *,
    ugid: str | None = None,
) -> None:
    """Use `tar` to copy a file or directory into the container.

    The files created within the schroot environment will be owned by the user account that
    the current process is running under.
    """
    self.cfg.log.info(
        "Copying %(src)s into the %(sess)s session under %(dst)s",
        {"src": src, "sess": self.sess, "dst": dst_parent},
    )

    ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
    tar_extract: Final = subprocess.Popen(  # noqa: S603
        [
            self.cfg.schroot_cmd,
            "-r",
            "-c",
            self.sess,
            "-d",
            dst_parent,
            *ugid_cmd,
            "--",
            "tar",
            "xf",
            "-",
        ],
        stdin=subprocess.PIPE,
        shell=False,
    )
    self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

    tar_create: Final = subprocess.Popen(  # noqa: S603
        ["tar", "cf", "-", src.name],  # noqa: S607
        stdout=tar_extract.stdin,
        shell=False,
        cwd=src.parent,
    )
    self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_extract.pid})

    self.cfg.log.debug("- waiting for the create process to complete")
    rc_create: Final = tar_create.wait()
    self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

    self.cfg.log.debug("- waiting for the extract process to complete")
    rc_extract: Final = tar_extract.wait()
    self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

    if rc_create != 0 or rc_extract != 0:
        raise defs.TarFilesError(src, dst_parent)

tar_files_out(src, dst_parent, ugid=None)

Use tar to copy a file or directory out of the container.

The files created on the host filesystem will be owned by the user account that the current process is running under.

Source code in src/run_isolated/rschroot.py

def tar_files_out(
    self,
    src: pathlib.Path,
    dst_parent: pathlib.Path,
    ugid: str | None = None,
) -> None:
    """Use `tar` to copy a file or directory out of the container.

    The files created on the host filesystem will be owned by the user account that
    the current process is running under.
    """
    self.cfg.log.info(
        "Copying %(src)s out of the %(sess)s session under %(dst)s",
        {"src": src, "sess": self.sess, "dst": dst_parent},
    )

    tar_extract: Final = subprocess.Popen(  # noqa: S603
        ["tar", "xf", "-", src.name],  # noqa: S607
        stdin=subprocess.PIPE,
        shell=False,
        cwd=dst_parent,
    )
    self.cfg.log.debug("- spawned tar extract process %(pid)d", {"pid": tar_extract.pid})

    ugid_cmd: Final = ["-u", ugid] if ugid is not None else []
    tar_create: Final = subprocess.Popen(  # noqa: S603
        [
            self.cfg.schroot_cmd,
            "-r",
            "-c",
            self.sess,
            "-d",
            src.parent,
            *ugid_cmd,
            "--",
            "tar",
            "cf",
            "-",
            src.name,
        ],
        stdout=tar_extract.stdin,
        shell=False,
    )
    self.cfg.log.debug("- spawned tar create process %(pid)d", {"pid": tar_extract.pid})

    self.cfg.log.debug("- waiting for the create process to complete")
    rc_create: Final = tar_create.wait()
    self.cfg.log.debug("- got create result %(rc)d", {"rc": rc_create})

    self.cfg.log.debug("- waiting for the extract process to complete")
    rc_extract: Final = tar_extract.wait()
    self.cfg.log.debug("- got extract result %(rc)d", {"rc": rc_extract})

    if rc_create != 0 or rc_extract != 0:
        raise defs.TarFilesError(src, dst_parent)

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