Skip to content

container

container

Core Container functionality for managing OCI images.

Container 

Container(*, image: str, user: int | None = None, group: int | None = None, mounts: list[tuple[PathOrStr, PathOrStr]] | None = None, cwd: PathOrStr | None = '/tmp', engine: ContainerEngine = 'docker', name: str | None = None, stdout: T.IO[bytes] | T.IO[str] | None = None, output_path: Path | None = None, logs_path: PathOrStr | None = None, additional_options: list[str] | None = None)

An object that represents a running OCI container.

Source code in src/mapyde/container.py 
def __init__(
    self,
    *,
    image: str,
    user: int | None = None,
    group: int | None = None,
    mounts: list[tuple[PathOrStr, PathOrStr]] | None = None,
    cwd: PathOrStr | None = "/tmp",
    engine: ContainerEngine = "docker",
    name: str | None = None,
    stdout: T.IO[bytes] | T.IO[str] | None = None,
    output_path: Path | None = None,
    logs_path: PathOrStr | None = None,
    additional_options: list[str] | None = None,
):
    if not image:
        msg = "Must specify an image to run."
        raise ValueError(msg)

    try:
        subprocess.run(["bash", "-c", f"hash {engine}"], check=True)
    except subprocess.CalledProcessError as err:
        msg = f"{engine} does not exist on your system."
        raise OSError(msg) from err

    self.image = image
    self.user = user or os.geteuid()
    self.group = group or os.getegid()
    self.mounts = mounts or []
    self.cwd = cwd
    self.engine = engine
    self.name = name
    self.stdin_config = subprocess.PIPE
    self.stdout_config = stdout or subprocess.PIPE
    self.stderr_config = subprocess.STDOUT
    self.output_path = (output_path or Path()).resolve()
    self.logs_path = logs_path
    self.additional_options = additional_options or []

    self.output_path.mkdir(parents=True, exist_ok=True)
    for host, container in self.mounts:
        if not Path(container).is_absolute():
            msg = f"The mount {host}:{container} does not point to an absolute path in the container."
            raise ValueError(msg)

call 

call(args: bytes, cwd: PathOrStr | None = None, env: dict[str, str] | None = None) -> tuple[bytes, bytes]

Execute the provided command in the container. A smarter version of Container.process.communicate(args).

Optionally change the current working directory (cwd) and set some environment variables.

Source code in src/mapyde/container.py 
def call(
    self,
    args: bytes,
    cwd: PathOrStr | None = None,
    env: dict[str, str] | None = None,
) -> tuple[bytes, bytes]:
    """
    Execute the provided command in the container. A smarter version of Container.process.communicate(args).

    Optionally change the current working directory (cwd) and set some environment variables.
    """
    if cwd is None:
        cwd = self.cwd

    chdir = ""
    if cwd:
        # singularity/apptainer mount host $TMPDIR into /tmp which might be
        # unexpectedly full of files so make an empty temporary directory
        chdir = "cd $(mktemp -d)" if cwd == "/tmp" else f"cd {cwd}"

    env_assignments = (
        " ".join(f"{shlex.quote(k)}={shlex.quote(v)}" for k, v in env.items())
        if env is not None
        else ""
    )

    command = bytes(
        f"""{chdir}
            env {env_assignments} {args.decode()}
            """,
        "utf-8",
    )

    return self.process.communicate(command)
Last update: June 15, 2023