docker

Docker image, container, orchestration, and cleanup helpers.

Examples:

>>> from ures.docker import Image
>>> image = Image("python", tag="3.12-slim")
>>> image.get_fullname()
'python:3.12-slim'

BuildConfig

Bases: BaseModel

BuildConfig defines the build parameters for constructing a Docker image.

Attributes:

• base_image (str) –

  Base image for the container. Default is "python:3.10-slim".

• platform (Optional[str]) –

  Target platform in format os[/arch[/variant]].

• python_deps_manager (Optional[str]) –

  Package manager for Python dependencies (e.g., pip, conda).

• python_dependencies (Optional[List[str]]) –

  List of Python packages to install.

• sys_deps_manager (Optional[str]) –

  Package manager for system dependencies (e.g., apt, yum, apk).

• sys_dependencies (Optional[List[str]]) –

  List of system packages to install.

• labels (Optional[List[Tuple[str, str]]]) –

  List of key-value tuples used as labels.

• uid (Optional[int]) –

  User ID to use inside the container.

• user (Optional[str]) –

  Username to use inside the container.

• entrypoint (Optional[List[str]]) –

  Entrypoint command for the container.

• cmd (Optional[List[str]]) –

  Default command to run in the container.

• environment (Optional[Dict[str, str]]) –

  Environment variables for the container.

• copies (Optional[List[Dict[str, str]]]) –

  File copy instructions (each dict should include "src" and "dest").

• context_dir (Union[str, Path]) –

  Directory for the build context. Defaults to the current working directory.

• docker_filename (str) –

  Filename for the Dockerfile. Defaults to "Dockerfile".

Examples:

>>> config = BuildConfig()
>>> config.base_image
'python:3.10-slim'

add_label

add_label(key: str, value: str)

Add a label to the build configuration.

Parameters:

• key (str) –

  The label key.

• value (str) –

  The label value.

Examples:

>>> config = BuildConfig()
>>> config.add_label("version", "1.0")
>>> config.labels
[("version", "1.0")]

Source code in ures/docker/conf.py

def add_label(self, key: str, value: str):
    """
    Add a label to the build configuration.

    Args:
        key (str): The label key.
        value (str): The label value.

    Examples:
        >>> config = BuildConfig()
        >>> config.add_label("version", "1.0")
        >>> config.labels
        [("version", "1.0")]
    """
    logger.info(f"Adding label {key} with value {value}")
    if self.labels is None:
        self.labels = []
    self.labels.append((key, value))

add_copy

add_copy(src: str, dest: str)

Add a file copy instruction to the build configuration.

Parameters:

• src (str) –

  Source file path.

• dest (str) –

  Destination path inside the container.

Examples:

>>> config = BuildConfig()
>>> config.add_copy("app.py", "/app/app.py")
>>> config.copies
[{"src": "app.py", "dest": "/app/app.py"}]

Source code in ures/docker/conf.py

def add_copy(self, src: str, dest: str):
    """
    Add a file copy instruction to the build configuration.

    Args:
        src (str): Source file path.
        dest (str): Destination path inside the container.

    Examples:
        >>> config = BuildConfig()
        >>> config.add_copy("app.py", "/app/app.py")
        >>> config.copies
        [{"src": "app.py", "dest": "/app/app.py"}]
    """
    logger.info(f"Adding copy {src} to {dest}")
    if self.copies is None:
        self.copies = []
    self.copies.append({"src": src, "dest": dest})

add_environment

add_environment(key: str, value: str)

Add an environment variable to the build configuration.

Parameters:

• key (str) –

  The environment variable name.

• value (str) –

  The value for the environment variable.

Examples:

>>> config = BuildConfig()
>>> config.add_environment("DEBUG", "true")
>>> config.environment["DEBUG"]
'true'

Source code in ures/docker/conf.py

def add_environment(self, key: str, value: str):
    """
    Add an environment variable to the build configuration.

    Args:
        key (str): The environment variable name.
        value (str): The value for the environment variable.

    Examples:
        >>> config = BuildConfig()
        >>> config.add_environment("DEBUG", "true")
        >>> config.environment["DEBUG"]
        'true'
    """
    logger.info(f"Adding environment variable {key} with value {value}")
    if self.environment is None:
        self.environment = {}
    self.environment[key] = value

set_context_dir

set_context_dir(context_dir: Union[str, Path])

Set the build context directory.

Parameters:

• context_dir (Union[str, Path]) –

  The directory to be used as the build context.

Raises:

• ValueError –

  If the provided context directory does not exist or is not a directory.

Examples:

>>> from pathlib import Path
>>> config = BuildConfig()
>>> temp_dir = Path("/tmp")
>>> config.set_context_dir(temp_dir)
>>> config.context_dir == temp_dir
True

Source code in ures/docker/conf.py

def set_context_dir(self, context_dir: Union[str, Path]):
    """
    Set the build context directory.

    Args:
        context_dir (Union[str, Path]): The directory to be used as the build context.

    Raises:
        ValueError: If the provided context directory does not exist or is not a directory.

    Examples:
        >>> from pathlib import Path
        >>> config = BuildConfig()
        >>> temp_dir = Path("/tmp")
        >>> config.set_context_dir(temp_dir)
        >>> config.context_dir == temp_dir
        True
    """
    logger.info(f"Setting context directory to {context_dir}")
    if isinstance(context_dir, str):
        context_dir = Path(context_dir)
    if not context_dir.is_dir():
        raise ValueError(f"Context directory {context_dir} is not a directory")
    self.context_dir = context_dir

add_python_dependency

add_python_dependency(dependency: str)

Add a Python dependency to be installed in the image.

Parameters:

• dependency (str) –

  The Python package dependency (e.g., "flask==2.0.1").

Examples:

>>> config = BuildConfig()
>>> config.add_python_dependency("flask")
>>> "flask" in config.python_dependencies
True

Source code in ures/docker/conf.py

def add_python_dependency(self, dependency: str):
    """
    Add a Python dependency to be installed in the image.

    Args:
        dependency (str): The Python package dependency (e.g., "flask==2.0.1").

    Examples:
        >>> config = BuildConfig()
        >>> config.add_python_dependency("flask")
        >>> "flask" in config.python_dependencies
        True
    """
    logger.info(f"Adding Python dependency: '{dependency}'")
    if self.python_dependencies is None:
        self.python_dependencies = []
    self.python_dependencies.append(dependency)

add_system_dependency

add_system_dependency(dependency: str)

Add a system dependency to be installed in the image.

Parameters:

• dependency (str) –

  The system package dependency (e.g., "curl").

Examples:

>>> config = BuildConfig()
>>> config.add_system_dependency("curl")
>>> "curl" in config.sys_dependencies
True

Source code in ures/docker/conf.py

def add_system_dependency(self, dependency: str):
    """
    Add a system dependency to be installed in the image.

    Args:
        dependency (str): The system package dependency (e.g., "curl").

    Examples:
        >>> config = BuildConfig()
        >>> config.add_system_dependency("curl")
        >>> "curl" in config.sys_dependencies
        True
    """
    logger.info(f"Adding system dependency: '{dependency}'")
    if self.sys_dependencies is None:
        self.sys_dependencies = []
    self.sys_dependencies.append(dependency)

add_run_command

add_run_command(command: str)

Add a command to be run during the build process.

Parameters:

• command (str) –

  The command to run (e.g., "apt-get update").

Examples:

>>> config = BuildConfig()
>>> config.add_run_command("apt-get update")
>>> "apt-get update" in config.run_commands
True

Source code in ures/docker/conf.py

def add_run_command(self, command: str):
    """
    Add a command to be run during the build process.

    Args:
        command (str): The command to run (e.g., "apt-get update").

    Examples:
        >>> config = BuildConfig()
        >>> config.add_run_command("apt-get update")
        >>> "apt-get update" in config.run_commands
        True
    """
    logger.info(f"Adding run command: '{command}'")
    if self.run_commands is None:
        self.run_commands = []
    self.run_commands.append(command)

set_entrypoint

set_entrypoint(entrypoint: Union[str, List[str]])

Set the entrypoint for the container.

Parameters:

• entrypoint (Union[str, List[str]]) –

  The entrypoint command(s). If a string is provided, it will be converted to a list.

Examples:

>>> config = BuildConfig()
>>> config.set_entrypoint("python app.py")
>>> config.entrypoint
["python app.py"]

Source code in ures/docker/conf.py

def set_entrypoint(self, entrypoint: Union[str, List[str]]):
    """
    Set the entrypoint for the container.

    Args:
        entrypoint (Union[str, List[str]]): The entrypoint command(s). If a string is provided,
                                            it will be converted to a list.

    Examples:
        >>> config = BuildConfig()
        >>> config.set_entrypoint("python app.py")
        >>> config.entrypoint
        ["python app.py"]
    """
    if isinstance(entrypoint, str):
        entrypoint = [entrypoint]
    logger.info(f"Setting entrypoint to: '{entrypoint}'")
    self.entrypoint = entrypoint

set_cmd

set_cmd(cmd: Union[str, List[str]])

Set the command for the container.

Parameters:

• cmd (Union[str, List[str]]) –

  The command(s) to run. If a string is provided, it will be converted to a list.

Examples:

>>> config = BuildConfig()
>>> config.set_cmd("python -m myapp")
>>> config.cmd
["python -m myapp"]

Source code in ures/docker/conf.py

def set_cmd(self, cmd: Union[str, List[str]]):
    """
    Set the command for the container.

    Args:
        cmd (Union[str, List[str]]): The command(s) to run. If a string is provided,
                                     it will be converted to a list.

    Examples:
        >>> config = BuildConfig()
        >>> config.set_cmd("python -m myapp")
        >>> config.cmd
        ["python -m myapp"]
    """
    if isinstance(cmd, str):
        cmd = [cmd]
    logger.info(f"Setting command to: '{cmd}'")
    self.cmd = cmd

RuntimeConfig

Bases: BaseModel

RuntimeConfig defines the runtime parameters for running a Docker container.

Attributes:

• image_name (str) –

  Name of the Docker image in the format "image:tag".

• name (Optional[str]) –

  Container name.

• platform (Optional[str]) –

  Platform for the container.

• detach (bool) –

  Whether to run the container in detached mode.

• user (Optional[str]) –

  User under which to run the container.

• remove (bool) –

  Whether to remove the container after it stops.

• cpus (Optional[int]) –

  Number of CPUs to allocate.

• gpus (Optional[List[str]]) –

  List of GPUs to allocate.

• gpu_driver (str) –

  Driver to use for GPUs. Default is "nvidia".

• memory (Optional[str]) –

  Memory limit for the container (e.g., "2g").

• entrypoint (Optional[List[Union[str, float, int, Path]]]) –

  Entrypoint command(s).

• command (Optional[List[Union[int, float, str, Path]]]) –

  Command(s) to run.

• env (Optional[Dict[str, str]]) –

  Environment variables.

• volumes (Optional[Dict[str, Dict[str, str]]]) –

  Volume mappings.

• subnet (Optional[str]) –

  Subnet for container networking.

• ipv4 (Optional[str]) –

  Specific IPv4 address for the container.

• subnet_mask (Optional[str]) –

  Subnet mask (e.g., "172.17.0.0/16").

• subnet_gateway (Optional[str]) –

  Subnet gateway.

• network_mode (Optional[str]) –

  Docker network mode.

• out_dir (Path) –

  Output directory for logs and cache.

Examples:

>>> config = RuntimeConfig()
>>> config.image_name
'model-runner'

log_dir property

log_dir: Path

Get the log directory within the output directory. If it does not exist, it is created.

Returns:

• Path ( Path ) –

  The path to the log directory.

Examples:

>>> config = RuntimeConfig()
>>> log_directory = config.log_dir
>>> log_directory.exists()
True

cache property

cache: Path

Get the cache directory within the output directory. If it does not exist, it is created.

Returns:

• Path ( Path ) –

  The path to the cache directory.

Examples:

>>> config = RuntimeConfig()
>>> cache_directory = config.cache
>>> cache_directory.exists()
True

add_volume

add_volume(host_path: Union[str, Path], container_path: Union[str, Path], mode: str = 'rw')

Add a volume mapping for the container.

Parameters:

• host_path (Union[str, Path]) –

  The path on the host machine.

• container_path (Union[str, Path]) –

  The destination path inside the container.

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

  The mode for the volume mapping (e.g., "rw" or "ro"). Defaults to "rw".

Examples:

>>> config = RuntimeConfig()
>>> config.add_volume("/host/data", "/container/data", mode="rw")
>>> "/host/data" in config.volumes
True

Source code in ures/docker/conf.py

def add_volume(
    self,
    host_path: Union[str, Path],
    container_path: Union[str, Path],
    mode: str = "rw",
):
    """
    Add a volume mapping for the container.

    Args:
        host_path (Union[str, Path]): The path on the host machine.
        container_path (Union[str, Path]): The destination path inside the container.
        mode (str, optional): The mode for the volume mapping (e.g., "rw" or "ro"). Defaults to "rw".

    Examples:
        >>> config = RuntimeConfig()
        >>> config.add_volume("/host/data", "/container/data", mode="rw")
        >>> "/host/data" in config.volumes
        True
    """
    logger.info(f"Adding volume {host_path} to {container_path} with mode {mode}")
    if self.volumes is None:
        self.volumes = {}
    self.volumes[str(host_path)] = {"bind": str(container_path), "mode": mode}

add_env

add_env(key: str, value: str)

Add an environment variable for the container.

Parameters:

• key (str) –

  The environment variable name.

• value (str) –

  The value for the environment variable.

Examples:

>>> config = RuntimeConfig()
>>> config.add_env("DEBUG", "1")
>>> config.env["DEBUG"]
'1'

Source code in ures/docker/conf.py

def add_env(self, key: str, value: str):
    """
    Add an environment variable for the container.

    Args:
        key (str): The environment variable name.
        value (str): The value for the environment variable.

    Examples:
        >>> config = RuntimeConfig()
        >>> config.add_env("DEBUG", "1")
        >>> config.env["DEBUG"]
        '1'
    """
    logger.info(f"Adding environment variable {key} with value {value}")
    if self.env is None:
        self.env = {}
    self.env[key] = value

Container

Container(image: Image, client: Optional[DockerClient] = None)

A class to manage a Docker container.

This class provides methods to create a container with specified runtime configurations, manage network connections, and control the container's lifecycle (start, stop, remove, logs, wait).

Attributes:

• _image (Image) –

  The Docker image object to be used.

• _client (DockerClient) –

  The Docker client instance.

• _container (Optional[Container]) –

  The underlying Docker container object.

Initialize a Container instance.

Parameters:

• image (Image) –

  The Image object that provides the Docker image details.

• client (Optional[DockerClient], default: None ) –

  An optional Docker client instance. If not provided, docker.from_env() is used.

Examples:

>>> from ures.docker.image import Image
>>> img = Image("myapp")
>>> container = Container(img)

Source code in ures/docker/container.py

def __init__(self, image: Image, client: Optional[docker.DockerClient] = None):
    """
    Initialize a Container instance.

    Args:
        image (Image): The Image object that provides the Docker image details.
        client (Optional[docker.DockerClient]): An optional Docker client instance. If not provided,
            docker.from_env() is used.

    Examples:
        >>> from ures.docker.image import Image
        >>> img = Image("myapp")
        >>> container = Container(img)
    """
    self._image: Image = image
    self._client: docker.DockerClient = client or docker.from_env()
    self._container: Optional[DockerContainer] = None

image_name property

image_name: str

Retrieve the full image name (including tag).

Returns:

• str ( str ) –

  The full image name.

Examples:

>>> container.image_name
'myapp:latest'

is_created property

is_created: bool

Check whether the container has been created.

Returns:

• bool ( bool ) –

  True if the container exists, False otherwise.

Examples:

>>> container.is_created
False

status property

status: str

Get the current status of the container.

Returns:

• str ( str ) –

  The container status. If not found, returns "removed".

Examples:

>>> status = container.status
>>> status in ["created", "running", "exited", "removed"]
True

exit_code property

exit_code: int | None

Retrieve the exit code of the container's last run.

Returns:

• int | None –

  The exit code if Docker still knows the container, otherwise None.

Examples:

>>> code = container.exit_code
>>> isinstance(code, int) or code is None
True

is_running property

is_running: bool

Check if the container is currently running.

Returns:

• bool ( bool ) –

  True if running, False otherwise.

Examples:

>>> container.is_running
True

create

create(config: RuntimeConfig, tag: Optional[str] = None)

Create a Docker container using the provided runtime configuration.

Parameters:

• config (RuntimeConfig) –

  The runtime configuration for the container.

• tag (Optional[str], default: None ) –

  An optional image tag override. Defaults to None.

Examples:

>>> container.create(runtime_config)
>>> container.is_created
True

Source code in ures/docker/container.py

def create(self, config: RuntimeConfig, tag: Optional[str] = None):
    """
    Create a Docker container using the provided runtime configuration.

    Args:
        config (RuntimeConfig): The runtime configuration for the container.
        tag (Optional[str]): An optional image tag override. Defaults to None.

    Examples:
        >>> container.create(runtime_config)
        >>> container.is_created
        True
    """
    if config.image_name != self.image_name or tag != self._image.tag:
        config.image_name = self._image.get_fullname(tag=tag)
        logger.warning(f"The image name is updated to {config.image_name}")
    run_params = self._construct_build_params(config)
    logger.debug(
        f"Create {self.image_name} with running Configuration: {json.dumps(run_params)}"
    )
    _container = self._client.containers.create(**run_params)
    self._connect_to_network(_container, config)
    self._container = _container

stop

stop()

Stop the running container.

Examples:

>>> container.stop()

Source code in ures/docker/container.py

@check_instance_variable("_container")
def stop(self):
    """
    Stop the running container.

    Examples:
        >>> container.stop()
    """
    logger.debug(f"Stopping container: {self.image_name}")
    self._container.stop()

remove

remove()

Remove the container.

Examples:

>>> container.remove()

Source code in ures/docker/container.py

@check_instance_variable("_container")
def remove(self):
    """
    Remove the container.

    Examples:
        >>> container.remove()
    """
    logger.debug(f"Removing container: {self.image_name}")
    self._container.remove()
    self._container = None

logs

logs() -> bytes

Retrieve logs from the container.

Returns:

• bytes –

  Raw log output from the Docker SDK.

Examples:

>>> logs = container.logs()
>>> isinstance(logs, bytes)
True

Source code in ures/docker/container.py

@check_instance_variable("_container")
def logs(self) -> bytes:
    """
    Retrieve logs from the container.

    Returns:
        Raw log output from the Docker SDK.

    Examples:
        >>> logs = container.logs()
        >>> isinstance(logs, bytes)
        True
    """
    logger.debug(f"Retrieving logs from container: {self.image_name}")
    return self._container.logs()

wait

wait()

Wait for the container to finish execution.

Blocks until the Docker container stops. The Docker SDK result is not returned; inspect exit_code afterwards if you need the status.

Examples:

>>> container.wait()

Source code in ures/docker/container.py

@check_instance_variable("_container")
def wait(self):
    """
    Wait for the container to finish execution.

    Blocks until the Docker container stops. The Docker SDK result is not
    returned; inspect ``exit_code`` afterwards if you need the status.

    Examples:
        >>> container.wait()
    """
    logger.debug(f"Waiting for container to finish: {self.image_name}")
    self._container.wait()

run

run()

Start the container if it has been created and is not already running.

Raises:

• RuntimeError –

  If the container has not been created or is already running.

Examples:

>>> container.run()
>>> container.is_running
True

Source code in ures/docker/container.py

def run(self):
    """
    Start the container if it has been created and is not already running.

    Raises:
        RuntimeError: If the container has not been created or is already running.

    Examples:
        >>> container.run()
        >>> container.is_running
        True
    """
    if self.is_created is True and self.is_running is False:
        self._container.start()
        logger.debug(f"Container started: {self.image_name}")
    else:
        if self.is_created is False:
            raise RuntimeError(f"Container has not been created: {self.image_name}")
        if self.is_running is True:
            raise RuntimeError(f"Container already running: {self.image_name}")

Containers

Containers(image: Image, client: Optional[DockerClient] = None, runtime: type(Runtime) = SimpleRuntime)

Manages the creation and lifecycle of Docker container instances derived from a given Docker image.

This class provides methods to create new container instances using a default or customized runtime configuration, run those containers using a specified runtime (default is SimpleRuntime), and then stop and remove them while maintaining a history of container records and their configurations.

Parameters:

• image (Image) –

  An instance of the Image class representing the Docker image to be used.

• client (Optional[DockerClient], default: None ) –

  A Docker client instance. Defaults to docker.from_env().

• runtime (type(Runtime), default: SimpleRuntime ) –

  The runtime class used to manage container operations. Must be a subclass of Runtime. Defaults to SimpleRuntime.

Attributes:

• _client (DockerClient) –

  The Docker client instance.

• _image (Image) –

  The Docker image used for creating containers.

• _runtime_history (Dict[str, Dict[str, Union[Container, str]]]) –

  A dictionary storing records of created container instances and their associated runtime configurations, keyed by a unique container name.

• _runtime (type(Runtime) –

  The runtime class used for managing container lifecycle operations.

Methods:

• get_container –

  bool = False) -> List[Tuple[str, Dict[str, Union[Container, str]]]]: Retrieve container records from the history. If new is True, returns only records whose container status is "created".

• _default_config –

  Generates a default RuntimeConfig based on the image and a generated unique container name.

• _construct_config –

  Constructs a RuntimeConfig by updating the default configuration with provided keyword arguments.

• _add_record –

  Container, config: RuntimeConfig) -> None: Adds a record for a newly created container and its configuration to the runtime history.

• create –

  Creates a new container using the provided configuration keyword arguments, records it in the history, and returns the container.

• run –

  Runs all newly created containers using the specified runtime and returns their records.

• stop –

  Optional[Union[str, Path]] = None) -> List[Tuple[str, Dict[str, Union[Container, str]]]]: Stops all managed containers, collects their logs (saving them to the specified directory), removes them, and returns the updated container records.

Examples:

>>> from ures.docker.image import Image
>>> from ures.docker.containers import Containers
>>> from ures.docker.conf import RuntimeConfig
>>> img = Image("myapp", tag="latest")
>>> containers_manager = Containers(image=img)
>>> container = containers_manager.create(name="myapp-instance")
>>> containers_manager.run()
>>> containers_manager.stop("/tmp/container_logs")

Source code in ures/docker/containers.py

def __init__(
    self,
    image: Image,
    client: Optional[docker.DockerClient] = None,
    runtime: type(Runtime) = SimpleRuntime,
):
    # Ensure that the image is an instance of the Image class
    assert isinstance(image, Image)
    # Ensure that the provided runtime is a subclass of Runtime
    assert isinstance(runtime, type(Runtime))
    self._client = client or docker.from_env()
    self._image = image
    self._runtime_history: Dict[str, Dict[str, Union[Container, str]]] = {}
    self._runtime: type(Runtime) = runtime

image property

image: str

Returns the full image name including the tag.

Returns:

• str ( str ) –

  The full image name.

Examples:

>>> containers_manager.image
'myapp:latest'

name property

name: str

Generates a unique container name.

Returns:

• str –

  A name of the form {image.name}-instance-{unique_id}.

Examples:

>>> containers_manager.name
'myapp-instance-abc123def4'

history property

history: Dict[str, Dict[str, Union[Container, str]]]

Retrieves the history of container records.

Returns:

• Dict[str, Dict[str, Union[Container, str]]] –

  Dict[str, Dict[str, Union[Container, str]]]: A dictionary of container records keyed by unique names.

Examples:

>>> containers_manager.history
{'myapp-instance-abc123def4': {'container': <Container object>, 'config': <RuntimeConfig>}}

get_container

get_container(new: bool = False) -> List[Tuple[str, Dict[str, Union[Container, str]]]]

Retrieve container records from the history.

Parameters:

• new (bool, default: False ) –

  If True, only return container records whose container status is "created". Defaults to False.

Returns:

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  List[Tuple[str, Dict[str, Union[Container, str]]]]: A list of tuples containing the container's unique name

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  and its record.

Examples:

>>> records = containers_manager.get_container(new=True)

Source code in ures/docker/containers.py

def get_container(
    self, new: bool = False
) -> List[Tuple[str, Dict[str, Union[Container, str]]]]:
    """
    Retrieve container records from the history.

    Args:
        new (bool, optional): If True, only return container records whose container status is "created".
            Defaults to False.

    Returns:
        List[Tuple[str, Dict[str, Union[Container, str]]]]: A list of tuples containing the container's unique name
        and its record.

    Examples:
        >>> records = containers_manager.get_container(new=True)
    """
    if new:
        containers = list(
            filter(
                lambda x: x[1]["container"].status == "created",
                self._runtime_history.items(),
            )
        )
    else:
        containers = list(self._runtime_history.items())
    return containers

create

create(**kwargs) -> Container

Create a new container instance using the specified configuration.

Parameters:

• **kwargs (Any, default: {} ) –

  RuntimeConfig fields to override, such as name or detach.

Returns:

• Container ( Container ) –

  The newly created container instance.

Examples:

>>> container = containers_manager.create(name="instance1")

Source code in ures/docker/containers.py

def create(self, **kwargs) -> Container:
    """
    Create a new container instance using the specified configuration.

    Args:
        **kwargs (Any): RuntimeConfig fields to override, such as name or detach.

    Returns:
        Container: The newly created container instance.

    Examples:
        >>> container = containers_manager.create(name="instance1")
    """
    _container = Container(image=self._image, client=self._client)
    _conf = self._construct_config(**kwargs)
    _container.create(config=_conf, tag=None)
    self._add_record(container=_container, config=_conf)
    return _container

run

run() -> List[Tuple[str, Dict[str, Union[Container, str]]]]

Run all newly created containers using the specified runtime.

Returns:

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  List[Tuple[str, Dict[str, Union[Container, str]]]]: The list of container records that were run.

Examples:

>>> records = containers_manager.run()

Source code in ures/docker/containers.py

def run(self) -> List[Tuple[str, Dict[str, Union[Container, str]]]]:
    """
    Run all newly created containers using the specified runtime.

    Returns:
        List[Tuple[str, Dict[str, Union[Container, str]]]]: The list of container records that were run.

    Examples:
        >>> records = containers_manager.run()
    """
    new_containers = self.get_container(new=True)
    containers = [_c[1]["container"] for _c in new_containers]
    _runtime = self._runtime(containers=containers)
    _runtime.run()
    return new_containers

stop

stop(log_dir: Optional[Union[str, Path]] = None) -> List[Tuple[str, Dict[str, Union[Container, str]]]]

Stop all managed containers, collect their logs (saving them to the specified directory), and remove them.

Parameters:

• log_dir (Optional[Union[str, Path]], default: None ) –

  The directory where container logs will be saved. If not provided, a default temporary directory is used.

Returns:

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  List[Tuple[str, Dict[str, Union[Container, str]]]]: The updated container records after stopping and removal.

Examples:

>>> records = containers_manager.stop("/tmp/container_logs")

Source code in ures/docker/containers.py

def stop(
    self, log_dir: Optional[Union[str, Path]] = None
) -> List[Tuple[str, Dict[str, Union[Container, str]]]]:
    """
    Stop all managed containers, collect their logs (saving them to the specified directory), and remove them.

    Args:
        log_dir (Optional[Union[str, Path]], optional): The directory where container logs will be saved.
            If not provided, a default temporary directory is used.

    Returns:
        List[Tuple[str, Dict[str, Union[Container, str]]]]: The updated container records after stopping and removal.

    Examples:
        >>> records = containers_manager.stop("/tmp/container_logs")
    """
    log_dir = log_dir or get_temp_dir_with_specific_path("container-logs")
    log_dir = Path(log_dir)
    containers = [_c[1]["container"] for _c in self.get_container()]
    _runtime = self._runtime(containers=containers)
    _runtime.stop()
    _runtime.logs(output_dir=log_dir)
    _runtime.remove()
    return self.get_container()

Image

Image(image_name: str, tag: Optional[str] = None, client: DockerClient = None)

Represents and manages a Docker image.

Attributes:

• _image_name (str) –

  The name of the Docker image.

• _tag (str) –

  The tag of the Docker image (default is "latest").

• _client (DockerClient) –

  The Docker client instance.

• _image (Optional[Image]) –

  The Docker image object if available.

Initializes an Image instance.

Parameters:

• image_name (str) –

  The name of the Docker image.

• tag (Optional[str], default: None ) –

  The image tag. Defaults to "latest" if not provided.

• client (Optional[DockerClient], default: None ) –

  The Docker client to use. Defaults to docker.from_env().

Examples:

>>> img = Image("myapp", tag="v1")

Source code in ures/docker/image.py

def __init__(
    self,
    image_name: str,
    tag: Optional[str] = None,
    client: docker.DockerClient = None,
):
    """
    Initializes an Image instance.

    Args:
        image_name (str): The name of the Docker image.
        tag (Optional[str], optional): The image tag. Defaults to "latest" if not provided.
        client (Optional[docker.DockerClient], optional): The Docker client to use. Defaults to docker.from_env().

    Examples:
        >>> img = Image("myapp", tag="v1")
    """
    self._image_name = image_name
    self._tag = tag or "latest"
    self._client = client or docker.from_env()
    self._image: Optional[DockerImage] = None

name property

name: str

Gets the name of the image.

Returns:

• str ( str ) –

  The image name.

Examples:

>>> img = Image("myapp")
>>> img.name
'myapp'

tag property

tag: str

Gets the tag of the image.

Returns:

• str ( str ) –

  The image tag.

Examples:

>>> img = Image("myapp", tag="v1")
>>> img.tag
'v1'

exist property

exist: bool

Checks if the image exists locally.

Returns:

• bool ( bool ) –

  True if the image exists, False otherwise.

Examples:

>>> img = Image("myapp")
>>> img.exist  # Depends on local Docker images

image property

image: Optional[Image]

Gets the Docker image object.

Returns:

• Optional[Image] –

  Optional[DockerImage]: The Docker image object if found, otherwise None.

Examples:

>>> img = Image("myapp")
>>> img.image  # Might return a DockerImage object if available

id property

id: str

Gets the unique ID of the Docker image.

Returns:

• str ( str ) –

  The Docker image ID.

Examples:

>>> img = Image("myapp")
>>> img.id
'sha256:...'

architecture property

architecture: str

Gets the architecture of the Docker image.

Returns:

• str ( str ) –

  The image architecture (e.g., 'amd64').

Examples:

>>> img = Image("myapp")
>>> img.architecture
'amd64'

image_size property

image_size: int

Gets the size of the Docker image in bytes.

Returns:

• int ( int ) –

  The size of the image in bytes.

Examples:

>>> img = Image("myapp")
>>> img.image_size
12345678

labels property

labels: dict

Gets the labels of the Docker image.

Returns:

• dict ( dict ) –

  A dictionary of image labels.

Examples:

>>> img = Image("myapp")
>>> img.labels
{'version': '1.0'}

get_fullname

get_fullname(tag: Optional[str] = None) -> str

Constructs the full image name including the tag.

Parameters:

• tag (Optional[str], default: None ) –

  The tag to use; if not provided, the instance's tag is used.

Returns:

• str ( str ) –

  The full image name in the format "name:tag".

Examples:

>>> img = Image("myapp", tag="v1")
>>> img.get_fullname()
'myapp:v1'

Source code in ures/docker/image.py

def get_fullname(self, tag: Optional[str] = None) -> str:
    """
    Constructs the full image name including the tag.

    Args:
        tag (Optional[str], optional): The tag to use; if not provided, the instance's tag is used.

    Returns:
        str: The full image name in the format "name:tag".

    Examples:
        >>> img = Image("myapp", tag="v1")
        >>> img.get_fullname()
        'myapp:v1'
    """
    if tag is None:
        tag = self._tag
    return f"{self._image_name}:{tag}"

get_image

get_image(tag: Optional[str] = None) -> Optional[DockerImage]

Retrieves the Docker image from the local repository.

Parameters:

• tag (Optional[str], default: None ) –

  The tag to use when retrieving the image. Defaults to the instance's tag.

Returns:

• Optional[Image] –

  Optional[DockerImage]: The Docker image if found; otherwise, None.

Examples:

>>> img = Image("myapp")
>>> image_obj = img.get_image()
>>> image_obj is not None  # Depends on local Docker images

Source code in ures/docker/image.py

def get_image(self, tag: Optional[str] = None) -> Optional[DockerImage]:
    """
    Retrieves the Docker image from the local repository.

    Args:
        tag (Optional[str], optional): The tag to use when retrieving the image. Defaults to the instance's tag.

    Returns:
        Optional[DockerImage]: The Docker image if found; otherwise, None.

    Examples:
        >>> img = Image("myapp")
        >>> image_obj = img.get_image()
        >>> image_obj is not None  # Depends on local Docker images
    """
    image_name = self.get_fullname(tag=tag)
    logger.info(f"Getting image {image_name}")
    try:
        image = self._client.images.get(image_name)
        if tag is None or tag == self._tag:
            self._image = image
        return image
    except docker.errors.ImageNotFound:
        return None
    except docker.errors.APIError as e:
        logger.error(f"Error accessing Docker API: {e}")
        return None

pull_image

pull_image(tag: Optional[str] = None) -> Optional[DockerImage]

Pulls the Docker image from a remote repository.

Parameters:

• tag (Optional[str], default: None ) –

  The tag to pull; defaults to the instance's tag if not provided.

Returns:

• Optional[Image] –

  Optional[DockerImage]: The pulled Docker image if successful; otherwise, None.

Examples:

>>> img = Image("myapp")
>>> pulled = img.pull_image()
>>> pulled is not None

Source code in ures/docker/image.py

def pull_image(self, tag: Optional[str] = None) -> Optional[DockerImage]:
    """
    Pulls the Docker image from a remote repository.

    Args:
        tag (Optional[str], optional): The tag to pull; defaults to the instance's tag if not provided.

    Returns:
        Optional[DockerImage]: The pulled Docker image if successful; otherwise, None.

    Examples:
        >>> img = Image("myapp")
        >>> pulled = img.pull_image()
        >>> pulled is not None
    """
    tag = tag or self._tag
    logger.info(f"Pulling image {self.get_fullname(tag=tag)}")
    try:
        image = self._client.images.pull(self._image_name, tag=tag)
    except docker.errors.APIError as e:
        logger.error(f"Error pulling image {self.get_fullname(tag=tag)}")
    else:
        self._image = image
        return image

build_image

build_image(build_config: BuildConfig, dest: Union[str, Path], build_context: Optional[Union[str, Path]] = None) -> DockerImage

Builds a Docker image using the specified build configuration.

Parameters:

• build_config (BuildConfig) –

  The configuration for building the image.

• dest (Union[str, Path]) –

  The destination path where the Dockerfile will be saved.

• build_context (Optional[Union[str, Path]], default: None ) –

  The build context directory. Defaults to build_config.context_dir.

Returns:

• DockerImage ( Image ) –

  The built Docker image.

Examples:

>>> build_config = BuildConfig()
>>> img = Image("myapp")
>>> built_img = img.build_image(build_config, "/tmp/dockerfile_dir")
>>> built_img is not None
True

Source code in ures/docker/image.py

def build_image(
    self,
    build_config: BuildConfig,
    dest: Union[str, Path],
    build_context: Optional[Union[str, Path]] = None,
) -> DockerImage:
    """
    Builds a Docker image using the specified build configuration.

    Args:
        build_config (BuildConfig): The configuration for building the image.
        dest (Union[str, Path]): The destination path where the Dockerfile will be saved.
        build_context (Optional[Union[str, Path]], optional): The build context directory. Defaults to build_config.context_dir.

    Returns:
        DockerImage: The built Docker image.

    Examples:
        >>> build_config = BuildConfig()
        >>> img = Image("myapp")
        >>> built_img = img.build_image(build_config, "/tmp/dockerfile_dir")
        >>> built_img is not None
        True
    """
    build_context = build_context or build_config.context_dir
    build_context = Path(build_context)
    dest = Path(dest)
    if dest.is_dir():
        dest = dest.joinpath(build_config.docker_filename)
    builder = ImageConstructor(build_config)
    docker_path = builder.save(dest)
    image_name = self.get_fullname()
    args = {
        "path": str(build_context),
        "tag": image_name,
        "dockerfile": str(docker_path),
        "nocache": True,
    }
    logger.info(
        f"Building image {image_name} with dockerfile {docker_path} in context {build_context}"
    )
    try:
        image, build_log = self._client.images.build(**args)
    except docker.errors.BuildError as e:
        logger.error(f"Failed to build image {image_name}")
        for log in e.build_log:
            logger.error(log)
        raise e
    else:
        logger.info(f"Image {image_name} built successfully!")
        self._image = image
    for line in build_log:
        logger.debug(line)
    return image

remove

remove(tag: Optional[str] = None, force: bool = False, noprune: bool = False)

Removes the Docker image from the local repository.

Parameters:

• tag (Optional[str], default: None ) –

  The tag to remove. Defaults to the instance's tag.

• force (bool, default: False ) –

  Force removal. Defaults to False.

• noprune (bool, default: False ) –

  Do not remove untagged parent images. Defaults to False.

Examples:

>>> img = Image("myapp")
>>> img.remove()

Source code in ures/docker/image.py

def remove(
    self, tag: Optional[str] = None, force: bool = False, noprune: bool = False
):
    """
    Removes the Docker image from the local repository.

    Args:
        tag (Optional[str], optional): The tag to remove. Defaults to the instance's tag.
        force (bool, optional): Force removal. Defaults to False.
        noprune (bool, optional): Do not remove untagged parent images. Defaults to False.

    Examples:
        >>> img = Image("myapp")
        >>> img.remove()
    """
    image_name = self.get_fullname(tag=tag)
    args = {"image": image_name, "force": force, "noprune": noprune}
    try:
        logger.info(f"Removing image {image_name} with {args}")
        self._client.images.remove(**args)
    except docker.errors.APIError as e:
        logger.error(f"Failed to remove image {image_name}. Msg: {e}")
    finally:
        if self.exist:
            logger.error(f"Removing image {image_name} failed")
        else:
            logger.info(f"Removing image {image_name} succeeded")

info

info()

Prints detailed information about the Docker image.

Examples:

>>> img = Image("myapp")
>>> img.info()

Source code in ures/docker/image.py

def info(self):
    """
    Prints detailed information about the Docker image.

    Examples:
        >>> img = Image("myapp")
        >>> img.info()
    """
    print(
        "\033[1;33m====================================== Image Info ===============================================\033[0m"
    )
    print(f"Name: {self.name}")
    print(f"Image ID: {self.id}")
    print(f"Architecture: {self.architecture}")
    print(f"Image Size: {format_memory(self.image_size)}")
    print(f"Labels: {self.labels}")

ImageOrchestrator

ImageOrchestrator(client: Optional[DockerClient] = None)

Orchestrates the building of multiple Docker images considering their dependencies.

Attributes:

• _client (DockerClient) –

  The Docker client instance.

• _images (dict) –

  A dictionary holding images and their build configuration and status.

Initializes the ImageOrchestrator.

Parameters:

• client (Optional[DockerClient], default: None ) –

  The Docker client instance. Defaults to docker.from_env().

Examples:

>>> orchestrator = ImageOrchestrator()

Source code in ures/docker/image.py

def __init__(self, client: Optional[docker.DockerClient] = None):
    """
    Initializes the ImageOrchestrator.

    Args:
        client (Optional[docker.DockerClient], optional): The Docker client instance. Defaults to docker.from_env().

    Examples:
        >>> orchestrator = ImageOrchestrator()
    """
    self._client = client or docker.from_env()
    self._images: Dict[str, Dict[str, Union[Optional[Image], BuildConfig, str]]] = (
        {}
    )

images property

images: Dict[str, Dict[str, Union[Optional[Image], BuildConfig, str]]]

Retrieves the dictionary of managed images.

Returns:

• dict ( Dict[str, Dict[str, Union[Optional[Image], BuildConfig, str]]] ) –

  A dictionary mapping image fullnames to their configuration and status.

Examples:

>>> orch = ImageOrchestrator()
>>> orch.images  # Initially empty dictionary

add_image

add_image(image: Image, config: BuildConfig, base: Image = None) -> bool

Adds an image and its build configuration to the orchestrator.

Parameters:

• image (Image) –

  The Image instance to add.

• config (BuildConfig) –

  The build configuration for the image.

• base (Image, default: None ) –

  The base image that this image depends on, if any.

Returns:

• bool –

  True if the image was registered in the orchestrator.

Examples:

>>> orch = ImageOrchestrator()
>>> img = Image("myapp")
>>> config = BuildConfig()
>>> orch.add_image(img, config)
True

Source code in ures/docker/image.py

def add_image(self, image: Image, config: BuildConfig, base: Image = None) -> bool:
    """
    Adds an image and its build configuration to the orchestrator.

    Args:
        image (Image): The Image instance to add.
        config (BuildConfig): The build configuration for the image.
        base (Image, optional): The base image that this image depends on, if any.

    Returns:
        True if the image was registered in the orchestrator.

    Examples:
        >>> orch = ImageOrchestrator()
        >>> img = Image("myapp")
        >>> config = BuildConfig()
        >>> orch.add_image(img, config)
        True
    """
    assert isinstance(image, Image)
    assert isinstance(config, BuildConfig)
    assert image.get_image() not in self.images.keys()
    logger.info(f"Adding image {image.get_fullname()} to orchestrator")
    logger.info(f"{image.get_fullname()} image config: {config}")
    if base:
        assert isinstance(base, Image)
        assert base.get_fullname() in self.images.keys()
        logger.info(
            f"The image {image.get_fullname()} depends on base image {base.get_fullname()}"
        )
    self._images[image.get_fullname()] = {
        "image": image,
        "config": config,
        "base": base,
        "status": "init",
    }
    return image.get_fullname() in self._images.keys()

build_all

build_all()

Builds all registered images in the correct order based on dependencies.

Examples:

>>> orchestrator.build_all()

Source code in ures/docker/image.py

def build_all(self):
    """
    Builds all registered images in the correct order based on dependencies.

    Examples:
        >>> orchestrator.build_all()
    """
    build_sorted_list = self._topological_sort()
    tmp_dir = Path(
        get_temp_dir_with_specific_path(f"Bulk-Image-Build-{unique_id()}")
    )
    logger.info(f"Building images in temporary directory: {tmp_dir}")
    for image_key in tqdm(build_sorted_list):
        image: Image = self._images[image_key]["image"]
        config: BuildConfig = self._images[image_key]["config"]
        base: Optional[Image] = self._images[image_key]["base"]
        logger.debug(f"starting build for {image.get_fullname()}")
        if base is not None:
            logger.debug(f"Original config: {config}")
            base_config: BuildConfig = self._images[base.get_fullname()]["config"]
            config.base_image = base.get_fullname()
            config.python_deps_manager = base_config.python_deps_manager
            config.sys_deps_manager = base_config.sys_deps_manager
            config.user = base_config.user
            config.uid = base_config.uid
            config.add_label("BaseImage", base.get_fullname())
            logger.debug(f"Config after inheritance: {config}")

        target_dir = tmp_dir.joinpath(image.get_fullname().replace(":", "-"))
        logger.info(f"The Dockerfile will be saved to {target_dir}")
        image.build_image(build_config=config, dest=target_dir)
        if not image.exist:
            self.images[image_key]["status"] = "failed"
            raise RuntimeError(f"Failed to build image {image.get_fullname()}")
        else:
            self._images[image_key]["status"] = "success"

DockerCleanup

DockerCleanup(client: Optional[DockerClient] = None)

Remove dangling images and leftover containers from the local Docker engine.

Examples:

>>> from ures.docker import DockerCleanup
>>> cleanup = DockerCleanup()
>>> cleanup.dangling_images()
>>> cleanup.stopped_containers()

Attach to a Docker client.

Parameters:

• client (DockerClient | None, default: None ) –

  A Docker SDK client. Defaults to docker.from_env().

Source code in ures/docker/cleanup.py

def __init__(self, client: Optional[docker.DockerClient] = None):
    """Attach to a Docker client.

    Args:
        client (docker.DockerClient | None): A Docker SDK client. Defaults to
            ``docker.from_env()``.
    """
    self.client = client or docker.from_env()

dangling_images

dangling_images()

Delete dangling (untagged) local images.

Source code in ures/docker/cleanup.py

def dangling_images(self):
    """Delete dangling (untagged) local images."""
    client = self.client

    # List dangling images (dangling=true filter)
    dangling_images = client.images.list(filters={"dangling": True})

    if not dangling_images:
        print("No dangling images to remove.")
        return

    for image in dangling_images:
        try:
            print(f"Removing image: {image.id}")
            client.images.remove(image.id)
        except Exception as e:
            print(f"Error removing image {image.id}: {e}")

    print("Dangling images prune complete!")

stopped_containers

stopped_containers()

Delete containers in exited or created state.

Source code in ures/docker/cleanup.py

def stopped_containers(self):
    """Delete containers in ``exited`` or ``created`` state."""
    client = self.client

    stopped_containers = client.containers.list(
        all=True, filters={"status": "exited"}
    )
    just_created_containers = client.containers.list(
        all=True, filters={"status": "created"}
    )

    plan2remove_containers = stopped_containers + just_created_containers

    if not plan2remove_containers:
        print("No stopped xmem_container to prune.")
        return

    for container in plan2remove_containers:
        try:
            print(f"Removing container: {container.name} ({container.short_id})")
            container.remove()
        except Exception as e:
            print(f"Error removing container {container.name}: {e}")

    print("Pruning complete!")