Docker Wrapper

Container

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:

Name	Type	Description
_image	Image	The Docker image object to be used.
_client	DockerClient	The Docker client instance.
_container	Optional[Container]	The underlying Docker container object.

Source code in ures/docker/container.py

class Container:
    """
    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 (docker.DockerClient): The Docker client instance.
        _container (Optional[DockerContainer]): The underlying Docker container object.
    """

    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.

        Example:
            >>> 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

    @property
    def image_name(self) -> str:
        """
        Retrieve the full image name (including tag).

        Returns:
            str: The full image name.

        Example:
            >>> container.image_name
            'myapp:latest'
        """
        return self._image.get_fullname()

    @property
    def is_created(self) -> bool:
        """
        Check whether the container has been created.

        Returns:
            bool: True if the container exists, False otherwise.

        Example:
            >>> container.is_created
            False
        """
        return self._container is not None

    @property
    def status(self) -> str:
        """
        Get the current status of the container.

        Returns:
            str: The container status. If not found, returns "removed".

        Example:
            >>> status = container.status
            >>> status in ["created", "running", "exited", "removed"]
            True
        """
        try:
            status = self._client.containers.get(self._container.id).status
        except docker.errors.NotFound:
            status = "removed"
        return status

    @property
    def exit_code(self):
        """
        Retrieve the exit code of the container's last run.

        Returns:
            int or None: The exit code if available, otherwise None.

        Example:
            >>> code = container.exit_code
            >>> isinstance(code, int) or code is None
            True
        """
        try:
            exit_code = self._client.containers.get(self._container.id).attrs["State"][
                "ExitCode"
            ]
        except docker.errors.NotFound:
            exit_code = None
        return exit_code

    @property
    def is_running(self) -> bool:
        """
        Check if the container is currently running.

        Returns:
            bool: True if running, False otherwise.

        Example:
            >>> container.is_running
            True
        """
        return self.status == "running"

    def _construct_build_params(self, config: RuntimeConfig) -> dict:
        """
        Construct runtime parameters for creating the container from the given configuration.

        Args:
            config (RuntimeConfig): The runtime configuration for the container.

        Returns:
            dict: A dictionary of parameters to be passed to Docker for container creation.

        Example:
            >>> params = container._construct_build_params(config)
            >>> isinstance(params, dict)
            True
        """
        _config: RuntimeConfig = config
        _params = {
            "image": _config.image_name,
            "auto_remove": _config.remove,
            "detach": _config.detach,
        }
        logger.debug(f"Creating container with configuration: {json.dumps(_params)}")
        if _config.cpus is not None:
            logger.debug(f"Setting cpuset_cpus to {_config.cpus}")
            _params["cpuset_cpus"] = _config.cpus
        if _config.gpus is not None:
            logger.debug(f"Setting device_requests to {_config.gpus}")
            _params["device_requests"] = [
                docker.types.DeviceRequest(
                    driver=_config.gpu_driver,
                    device_ids=_config.gpus,
                    capabilities=[["gpu"]],
                )
            ]
        if _config.memory is not None:
            logger.debug(f"Setting mem_limit to {_config.memory}")
            _params["mem_limit"] = _config.memory
        if _config.entrypoint is not None:
            logger.debug(f"Setting entrypoint to {_config.entrypoint}")
            _params["entrypoint"] = [str(entry) for entry in _config.entrypoint]
        if _config.command is not None:
            logger.debug(f"Setting command to {_config.command}")
            _params["command"] = [str(cmd) for cmd in _config.command]
        if _config.volumes is not None:
            logger.debug(f"Setting volumes to {_config.volumes}")
            _params["volumes"] = _config.volumes
        if _config.env is not None:
            logger.debug(f"Setting environment to {_config.env}")
            _params["environment"] = _config.env
        if _config.name is not None:
            logger.debug(f"Setting name to {_config.name}")
            _params["name"] = _config.name
        if _config.user is not None:
            logger.debug(f"Setting user to {_config.user}")
            _params["user"] = _config.user
        return _params

    def _create_subnet(self, config: RuntimeConfig) -> docker.models.networks.Network:
        """
        Create a new Docker network (subnet) based on the provided configuration.

        Args:
            config (RuntimeConfig): The runtime configuration containing subnet parameters.

        Returns:
            docker.models.networks.Network: The created Docker network.

        Raises:
            RuntimeError: If the subnet creation fails.

        Example:
            >>> net = container._create_subnet(config)
            >>> net.name == config.subnet
            True
        """
        submask_list = config.subnet_mask.split("/")
        assert is_valid_ip_netmask(ip=submask_list[0], netmask=submask_list[1])
        assert verify_ip_in_subnet(ip=config.subnet_gateway, subnet=config.subnet_mask)
        try:
            logger.debug(
                f"Creating IPAM for docker network with mask {config.subnet_mask} and gateway {config.subnet_gateway}"
            )
            logger.debug(
                f"Create network with name {config.subnet}, driver {config.network_mode}"
            )
            ipam_pool = docker.types.IPAMPool(
                subnet=config.subnet_mask, gateway=config.subnet_gateway
            )
            ipam_config = docker.types.IPAMConfig(pool_configs=[ipam_pool])
            network = self._client.networks.create(
                **{
                    "name": config.subnet,
                    "driver": config.network_mode,
                    "ipam": ipam_config,
                }
            )
        except docker.errors.APIError as e:
            raise RuntimeError(f"Failed to create subnet {config.subnet}") from e
        else:
            return network

    def _connect_to_network(self, contain: DockerContainer, config: RuntimeConfig):
        """
        Connect the container to a Docker network based on the provided configuration.

        Args:
            contain (DockerContainer): The Docker container object to connect.
            config (RuntimeConfig): The runtime configuration with network details.

        Returns:
            None

        Example:
            >>> container._connect_to_network(docker_container, config)
        """
        if config.subnet is not None:
            try:
                logger.info(f"Connecting container to network: {config.subnet}")
                net = self._client.networks.get(config.subnet)
            except docker.errors.NotFound as e:
                logger.warning(f"Could not find network: {config.subnet} with msg {e}")
                net = self._create_subnet(config=config)
            finally:
                verify_ip_in_subnet(
                    ip=config.ipv4, subnet=net.attrs["IPAM"]["Config"][0]["Subnet"]
                )
                logger.debug(f"Connecting container to network with IP {config.ipv4}")
                net.connect(contain, ipv4_address=config.ipv4)

    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.

        Returns:
            None

        Example:
            >>> 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

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

        Returns:
            None

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

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

        Returns:
            None

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

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

        Returns:
            bytes: The log output from the container.

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

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

        Returns:
            dict: The container's exit information.

        Example:
            >>> exit_info = container.wait()
            >>> "StatusCode" in exit_info
            True
        """
        logger.debug(f"Waiting for container to finish: {self.image_name}")
        self._container.wait()

    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.

        Returns:
            None

        Example:
            >>> 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}")

exit_code property

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

Returns:

Type	Description
	int or None: The exit code if available, otherwise None.

Example

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

image_name property

Retrieve the full image name (including tag).

Returns:

Name	Type	Description
str	str	The full image name.

Example

container.image_name 'myapp:latest'

is_created property

Check whether the container has been created.

Returns:

Name	Type	Description
bool	bool	True if the container exists, False otherwise.

Example

container.is_created False

is_running property

Check if the container is currently running.

Returns:

Name	Type	Description
bool	bool	True if running, False otherwise.

Example

container.is_running True

status property

Get the current status of the container.

Returns:

Name	Type	Description
str	str	The container status. If not found, returns "removed".

Example

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

__init__(image, client=None)

Initialize a Container instance.

Parameters:

Name	Type	Description	Default
image	Image	The Image object that provides the Docker image details.	required
client	Optional[DockerClient]	An optional Docker client instance. If not provided, docker.from_env() is used.	None

Example

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.

    Example:
        >>> 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

create(config, tag=None)

Create a Docker container using the provided runtime configuration.

Parameters:

Name	Type	Description	Default
config	RuntimeConfig	The runtime configuration for the container.	required
tag	Optional[str]	An optional image tag override. Defaults to None.	None

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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

logs()

Retrieve logs from the container.

Returns:

Name	Type	Description
bytes		The log output from the container.

Example

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

Source code in ures/docker/container.py

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

    Returns:
        bytes: The log output from the container.

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

remove()

Remove the container.

Returns:

Type	Description
	None

Example

container.remove()

Source code in ures/docker/container.py

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

    Returns:
        None

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

run()

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

Raises:

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

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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}")

stop()

Stop the running container.

Returns:

Type	Description
	None

Example

container.stop()

Source code in ures/docker/container.py

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

    Returns:
        None

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

wait()

Wait for the container to finish execution.

Returns:

Name	Type	Description
dict		The container's exit information.

Example

exit_info = container.wait() "StatusCode" in exit_info True

Source code in ures/docker/container.py

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

    Returns:
        dict: The container's exit information.

    Example:
        >>> exit_info = container.wait()
        >>> "StatusCode" in exit_info
        True
    """
    logger.debug(f"Waiting for container to finish: {self.image_name}")
    self._container.wait()

Image

Represents and manages a Docker image.

Attributes:

Name	Type	Description
_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.

Source code in ures/docker/image.py

class Image:
    """
    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 (docker.DockerClient): The Docker client instance.
        _image (Optional[DockerImage]): The Docker image object if available.
    """

    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().

        Example:
            >>> 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

    @property
    def name(self) -> str:
        """
        Gets the name of the image.

        Returns:
            str: The image name.

        Example:
            >>> img = Image("myapp")
            >>> img.name
            'myapp'
        """
        return self._image_name

    @property
    def tag(self) -> str:
        """
        Gets the tag of the image.

        Returns:
            str: The image tag.

        Example:
            >>> img = Image("myapp", tag="v1")
            >>> img.tag
            'v1'
        """
        return self._tag

    @property
    def exist(self) -> bool:
        """
        Checks if the image exists locally.

        Returns:
            bool: True if the image exists, False otherwise.

        Example:
            >>> img = Image("myapp")
            >>> img.exist  # Depends on local Docker images
        """
        return self.get_image() is not None

    @property
    def image(self) -> Optional[DockerImage]:
        """
        Gets the Docker image object.

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

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

    @property
    @check_instance_variable("image")
    def id(self) -> str:
        """
        Gets the unique ID of the Docker image.

        Returns:
            str: The Docker image ID.

        Example:
            >>> img = Image("myapp")
            >>> img.id
            'sha256:...'
        """
        return self.image.id

    @property
    @check_instance_variable("image")
    def architecture(self) -> str:
        """
        Gets the architecture of the Docker image.

        Returns:
            str: The image architecture (e.g., 'amd64').

        Example:
            >>> img = Image("myapp")
            >>> img.architecture
            'amd64'
        """
        return self.image.attrs["Architecture"]

    @property
    @check_instance_variable("image")
    def image_size(self) -> int:
        """
        Gets the size of the Docker image in bytes.

        Returns:
            int: The size of the image in bytes.

        Example:
            >>> img = Image("myapp")
            >>> img.image_size
            12345678
        """
        return int(self.image.attrs["Size"])

    @property
    @check_instance_variable("image")
    def labels(self) -> dict:
        """
        Gets the labels of the Docker image.

        Returns:
            dict: A dictionary of image labels.

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

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

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

    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.

        Example:
            >>> 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

    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.

        Example:
            >>> 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

    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.

        Example:
            >>> 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

    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.

        Returns:
            None

        Example:
            >>> 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")

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

        Returns:
            None

        Example:
            >>> 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}")

architecture property

Gets the architecture of the Docker image.

Returns:

Name	Type	Description
str	str	The image architecture (e.g., 'amd64').

Example

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

exist property

Checks if the image exists locally.

Returns:

Name	Type	Description
bool	bool	True if the image exists, False otherwise.

Example

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

id property

Gets the unique ID of the Docker image.

Returns:

Name	Type	Description
str	str	The Docker image ID.

Example

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

image property

Gets the Docker image object.

Returns:

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

Example

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

image_size property

Gets the size of the Docker image in bytes.

Returns:

Name	Type	Description
int	int	The size of the image in bytes.

Example

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

labels property

Gets the labels of the Docker image.

Returns:

Name	Type	Description
dict	dict	A dictionary of image labels.

Example

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

name property

Gets the name of the image.

Returns:

Name	Type	Description
str	str	The image name.

Example

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

tag property

Gets the tag of the image.

Returns:

Name	Type	Description
str	str	The image tag.

Example

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

__init__(image_name, tag=None, client=None)

Initializes an Image instance.

Parameters:

Name	Type	Description	Default
image_name	str	The name of the Docker image.	required
tag	Optional[str]	The image tag. Defaults to "latest" if not provided.	None
client	Optional[DockerClient]	The Docker client to use. Defaults to docker.from_env().	None

Example

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().

    Example:
        >>> 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

build_image(build_config, dest, build_context=None)

Builds a Docker image using the specified build configuration.

Parameters:

Name	Type	Description	Default
build_config	BuildConfig	The configuration for building the image.	required
dest	Union[str, Path]	The destination path where the Dockerfile will be saved.	required
build_context	Optional[Union[str, Path]]	The build context directory. Defaults to build_config.context_dir.	None

Returns:

Name	Type	Description
DockerImage	Image	The built Docker image.

Example

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.

    Example:
        >>> 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

get_fullname(tag=None)

Constructs the full image name including the tag.

Parameters:

Name	Type	Description	Default
tag	Optional[str]	The tag to use; if not provided, the instance's tag is used.	None

Returns:

Name	Type	Description
str	str	The full image name in the format "name:tag".

Example

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

    Example:
        >>> 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(tag=None)

Retrieves the Docker image from the local repository.

Parameters:

Name	Type	Description	Default
tag	Optional[str]	The tag to use when retrieving the image. Defaults to the instance's tag.	None

Returns:

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

Example

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.

    Example:
        >>> 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

info()

Prints detailed information about the Docker image.

Returns:

Type	Description
	None

Example

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

Source code in ures/docker/image.py

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

    Returns:
        None

    Example:
        >>> 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}")

pull_image(tag=None)

Pulls the Docker image from a remote repository.

Parameters:

Name	Type	Description	Default
tag	Optional[str]	The tag to pull; defaults to the instance's tag if not provided.	None

Returns:

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

Example

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.

    Example:
        >>> 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

remove(tag=None, force=False, noprune=False)

Removes the Docker image from the local repository.

Parameters:

Name	Type	Description	Default
tag	Optional[str]	The tag to remove. Defaults to the instance's tag.	None
force	bool	Force removal. Defaults to False.	False
noprune	bool	Do not remove untagged parent images. Defaults to False.	False

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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")

ImageConstructor

Constructs a Dockerfile based on a provided build configuration.

This class generates Dockerfile content by appending commands derived from the build configuration and provides methods to save the generated Dockerfile.

Source code in ures/docker/image.py

class ImageConstructor:
    """
    Constructs a Dockerfile based on a provided build configuration.

    This class generates Dockerfile content by appending commands derived from the build
    configuration and provides methods to save the generated Dockerfile.
    """

    def __init__(self, config: BuildConfig):
        """
        Initializes the ImageConstructor with the given build configuration.

        Args:
            config (BuildConfig): The build configuration settings.

        Returns:
            None

        Example:
            >>> from ures.docker.conf import BuildConfig
            >>> config = BuildConfig(base_image="python:3.10-slim", user="appuser")
            >>> constructor = ImageConstructor(config)
        """
        self._config = config
        self._dockerfile_content: List[str] = []
        self._build_dockerfile()

    @property
    def home_dir(self) -> Path:
        """
        Returns the home directory path based on the configured user.

        Returns:
            Path: The home directory path (e.g. /home/{user} if user is specified, else /root).

        Example:
            >>> from ures.docker.conf import BuildConfig
            >>> config = BuildConfig(user="appuser")
            >>> constructor = ImageConstructor(config)
            >>> constructor.home_dir
            PosixPath('/home/appuser')
        """
        home_dir = f"/home/{self._config.user}" if self._config.user else "/root"
        return Path(home_dir)

    @property
    def content(self) -> List[str]:
        """
        Retrieves the generated Dockerfile content as a list of command strings.

        Returns:
            List[str]: The Dockerfile content lines.

        Example:
            >>> constructor = ImageConstructor(BuildConfig())
            >>> constructor.content  # Might include commands like 'FROM python:3.10-slim'
        """
        return self._dockerfile_content

    def _add_command(self, command: str):
        """
        Appends a command line to the Dockerfile content.

        Args:
            command (str): The Dockerfile command to add.

        Returns:
            None

        Example:
            >>> constructor = ImageConstructor(BuildConfig())
            >>> constructor._add_command("RUN echo Hello")
        """
        logger.debug(f"Adding command: {command}")
        self._dockerfile_content.append(command)

    def save(self, dest: Union[str, Path]) -> Path:
        """
        Saves the generated Dockerfile to the specified destination.

        If the destination is a directory, the Dockerfile will be named using the configuration's
        docker_filename and placed inside that directory.

        Args:
            dest (Union[str, Path]): The destination file path or directory.

        Returns:
            Path: The full path where the Dockerfile was saved.

        Example:
            >>> constructor = ImageConstructor(BuildConfig(docker_filename="Dockerfile"))
            >>> saved_path = constructor.save("/tmp")
            >>> saved_path.name  # Should be 'Dockerfile'
        """
        dest_path = Path(dest) if isinstance(dest, str) else dest
        if dest_path.is_dir():
            dest_path = dest_path / self._config.docker_filename
        logger.info(f"Saving Dockerfile to: {dest_path}")
        dest_path.parent.mkdir(parents=True, exist_ok=True)
        with open(dest_path, "w") as f:
            f.write("\n".join(self._dockerfile_content))
        return dest_path

    def _set_base_image(self):
        """
        Sets the base image in the Dockerfile.

        If a platform is specified in the configuration, it is included in the FROM command.

        Returns:
            None

        Example:
            >>> constructor = ImageConstructor(BuildConfig(base_image="python:3.10-slim"))
            >>> constructor._set_base_image()
        """
        base_image = self._config.base_image
        if self._config.platform:
            base_image = f"--platform={self._config.platform} {base_image}"
        self._add_command(f"FROM {base_image}")

    def _set_labels(self):
        """
        Adds LABEL commands to the Dockerfile for each label defined in the build configuration.

        Returns:
            None

        Example:
            >>> config = BuildConfig(labels=[("version", "1.0")])
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_labels()
        """
        if self._config.labels:
            for key, value in self._config.labels:
                self._add_command(f'LABEL "{key}"="{value}"')

    def _set_user_and_workdir(self):
        """
        Sets the user and working directory in the Dockerfile.

        This command sets arguments for HOME_DIR and USER_NAME, creates the user if UID is specified,
        and then sets the USER and WORKDIR.

        Returns:
            None

        Example:
            >>> config = BuildConfig(user="appuser", uid=1001)
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_user_and_workdir()
        """
        user = self._config.user
        uid = self._config.uid
        self._add_command(f"ARG HOME_DIR={self.home_dir}")
        if user:
            self._add_command(f"ARG USER_NAME={user}")
            if uid:
                self._add_command(f"ARG UID={uid}")
                self._add_command(
                    "RUN id -u $USER_NAME >/dev/null 2>&1 || useradd -m -u $UID -s /bin/bash -d $HOME_DIR $USER_NAME"
                )
                self._add_command(f"RUN chown -R $USER_NAME:$USER_NAME $HOME_DIR")
            self._add_command(f"USER $USER_NAME")
        self._add_command(f"WORKDIR $HOME_DIR")

    def _set_system_dependencies(self):
        """
        Installs system dependencies in the Docker image using the specified package manager.

        Returns:
            None

        Example:
            >>> config = BuildConfig(sys_dependencies=["curl"], sys_deps_manager="apt")
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_system_dependencies()
        """
        sys_deps = self._config.sys_dependencies
        manager = self._config.sys_deps_manager
        if sys_deps:
            deps_string = " ".join(sys_deps)
            command = f"RUN {manager} update && {manager} install -y {deps_string}"
            if manager == "apt":
                command += " && apt-get clean && rm -rf /var/lib/apt/lists/*"
            self._add_command(command)

    def _set_python_dependencies(self):
        """
        Installs Python dependencies in the Docker image using the specified Python package manager.

        Returns:
            None

        Example:
            >>> config = BuildConfig(python_dependencies=["flask"], python_deps_manager="pip")
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_python_dependencies()
        """
        python_deps = self._config.python_dependencies
        manager = self._config.python_deps_manager
        if python_deps:
            deps_string = " ".join(python_deps)
            if manager == "pip":
                command = (
                    f"RUN pip install --upgrade pip && pip install --no-cache-dir {deps_string} "
                    f"&& rm -rf /tmp/* /var/tmp/*"
                )
                self._add_command(command)
            elif manager == "conda":
                # Placeholder for conda support.
                pass
            else:
                logger.warning(f"Unsupported python package manager: {manager}")

    def _set_run_commands(self):
        """
        Adds RUN commands to the Dockerfile for each command in the build configuration.

        Returns:
            None

        Example:
            >>> config = BuildConfig(run_commands=["echo Hello"])
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_run_commands()
        """
        if self._config.run_commands:
            for command in self._config.run_commands:
                self._add_command(f"RUN {command}")

    def _set_copies(self):
        """
        Adds COPY commands to the Dockerfile for each file copy instruction in the build configuration.

        Returns:
            None

        Example:
            >>> config = BuildConfig(copies=[{"src": "app.py", "dest": "/app/app.py"}])
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_copies()
        """
        if self._config.copies:
            for copy_spec in self._config.copies:
                src = Path(copy_spec["src"])
                dest = Path(copy_spec["dest"])
                if not dest.is_absolute():
                    dest = self.home_dir.joinpath(dest)
                if self._config.user is None:
                    self._add_command(f"COPY {src} {dest}")
                else:
                    self._add_command(f"COPY --chown=$USER_NAME {src} {dest}")

    def _set_environment(self):
        """
        Sets environment variables in the Dockerfile using the ENV command.

        Returns:
            None

        Example:
            >>> config = BuildConfig(environment={"DEBUG": "true"})
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_environment()
        """
        if self._config.environment:
            for key, value in self._config.environment.items():
                self._add_command(f"ENV {key}={value}")

    def _set_entrypoint(self):
        """
        Sets the ENTRYPOINT in the Dockerfile using the build configuration.

        Returns:
            None

        Example:
            >>> config = BuildConfig(entrypoint=["python", "app.py"])
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_entrypoint()
        """
        if self._config.entrypoint:
            self._add_command(f"ENTRYPOINT {json.dumps(self._config.entrypoint)}")

    def _set_cmd(self):
        """
        Sets the CMD in the Dockerfile if no ENTRYPOINT is specified.

        Returns:
            None

        Example:
            >>> config = BuildConfig(cmd=["python", "-m", "app"])
            >>> constructor = ImageConstructor(config)
            >>> constructor._set_cmd()
        """
        if self._config.cmd and not self._config.entrypoint:
            self._add_command(f"CMD {json.dumps(self._config.cmd)}")

    def _build_dockerfile(self):
        """
        Builds the complete Dockerfile content based on the build configuration.

        Returns:
            None

        Example:
            >>> constructor = ImageConstructor(BuildConfig())
            >>> constructor.content  # Contains all Dockerfile commands
        """
        self._set_base_image()
        self._set_labels()
        self._set_system_dependencies()
        self._set_python_dependencies()
        self._set_run_commands()
        self._set_user_and_workdir()
        self._set_copies()
        self._set_environment()
        self._set_entrypoint()
        self._set_cmd()

content property

Retrieves the generated Dockerfile content as a list of command strings.

Returns:

Type	Description
List[str]	List[str]: The Dockerfile content lines.

Example

constructor = ImageConstructor(BuildConfig()) constructor.content # Might include commands like 'FROM python:3.10-slim'

home_dir property

Returns the home directory path based on the configured user.

Returns:

Name	Type	Description
Path	Path	The home directory path (e.g. /home/{user} if user is specified, else /root).

Example

from ures.docker.conf import BuildConfig config = BuildConfig(user="appuser") constructor = ImageConstructor(config) constructor.home_dir PosixPath('/home/appuser')

__init__(config)

Initializes the ImageConstructor with the given build configuration.

Parameters:

Name	Type	Description	Default
config	BuildConfig	The build configuration settings.	required

Returns:

Type	Description
	None

Example

from ures.docker.conf import BuildConfig config = BuildConfig(base_image="python:3.10-slim", user="appuser") constructor = ImageConstructor(config)

Source code in ures/docker/image.py

def __init__(self, config: BuildConfig):
    """
    Initializes the ImageConstructor with the given build configuration.

    Args:
        config (BuildConfig): The build configuration settings.

    Returns:
        None

    Example:
        >>> from ures.docker.conf import BuildConfig
        >>> config = BuildConfig(base_image="python:3.10-slim", user="appuser")
        >>> constructor = ImageConstructor(config)
    """
    self._config = config
    self._dockerfile_content: List[str] = []
    self._build_dockerfile()

save(dest)

Saves the generated Dockerfile to the specified destination.

If the destination is a directory, the Dockerfile will be named using the configuration's docker_filename and placed inside that directory.

Parameters:

Name	Type	Description	Default
dest	Union[str, Path]	The destination file path or directory.	required

Returns:

Name	Type	Description
Path	Path	The full path where the Dockerfile was saved.

Example

constructor = ImageConstructor(BuildConfig(docker_filename="Dockerfile")) saved_path = constructor.save("/tmp") saved_path.name # Should be 'Dockerfile'

Source code in ures/docker/image.py

def save(self, dest: Union[str, Path]) -> Path:
    """
    Saves the generated Dockerfile to the specified destination.

    If the destination is a directory, the Dockerfile will be named using the configuration's
    docker_filename and placed inside that directory.

    Args:
        dest (Union[str, Path]): The destination file path or directory.

    Returns:
        Path: The full path where the Dockerfile was saved.

    Example:
        >>> constructor = ImageConstructor(BuildConfig(docker_filename="Dockerfile"))
        >>> saved_path = constructor.save("/tmp")
        >>> saved_path.name  # Should be 'Dockerfile'
    """
    dest_path = Path(dest) if isinstance(dest, str) else dest
    if dest_path.is_dir():
        dest_path = dest_path / self._config.docker_filename
    logger.info(f"Saving Dockerfile to: {dest_path}")
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    with open(dest_path, "w") as f:
        f.write("\n".join(self._dockerfile_content))
    return dest_path

ImageOrchestrator

Orchestrates the building of multiple Docker images considering their dependencies.

Attributes:

Name	Type	Description
_client	DockerClient	The Docker client instance.
_images	dict	A dictionary holding images and their build configuration and status.

Source code in ures/docker/image.py

class ImageOrchestrator:
    """
    Orchestrates the building of multiple Docker images considering their dependencies.

    Attributes:
        _client (docker.DockerClient): The Docker client instance.
        _images (dict): A dictionary holding images and their build configuration and status.
    """

    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().

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

    @property
    def images(self) -> Dict[str, Dict[str, Union[Optional[Image], BuildConfig, str]]]:
        """
        Retrieves the dictionary of managed images.

        Returns:
            dict: A dictionary mapping image fullnames to their configuration and status.

        Example:
            >>> orch = ImageOrchestrator()
            >>> orch.images  # Initially empty dictionary
        """
        return self._images

    def add_image(self, image: Image, config: BuildConfig, base: Image = None):
        """
        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:
            bool: True if the image was added successfully.

        Example:
            >>> 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()

    def _topological_sort(self) -> list:
        """
        Performs a topological sort of images based on their dependency relationships.

        Returns:
            list: A list of image fullnames in the order they should be built.

        Raises:
            Exception: If a circular dependency is detected or a base image is not registered.

        Example:
            >>> sorted_images = orchestrator._topological_sort()
        """
        sorted_list = []
        temporary_marks = set()
        permanent_marks = set()

        def visit(image_key: str) -> None:
            if image_key in permanent_marks:
                return
            if image_key in temporary_marks:
                raise Exception(f"Circular dependency detected: {image_key}")
            temporary_marks.add(image_key)
            base = self._images[image_key].get("base")
            if base:
                base_key = base.get_fullname()
                if base_key in self._images:
                    visit(base_key)
                else:
                    raise Exception(f"Base image {base_key} is not registered")
            permanent_marks.add(image_key)
            sorted_list.append(image_key)

        for key in self._images:
            if key not in permanent_marks:
                visit(key)
        logger.debug(f"After topological sort: {sorted_list}")
        return sorted_list

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

        Returns:
            None

        Example:
            >>> 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"

images property

Retrieves the dictionary of managed images.

Returns:

Name	Type	Description
dict	Dict[str, Dict[str, Union[Optional[Image], BuildConfig, str]]]	A dictionary mapping image fullnames to their configuration and status.

Example

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

__init__(client=None)

Initializes the ImageOrchestrator.

Parameters:

Name	Type	Description	Default
client	Optional[DockerClient]	The Docker client instance. Defaults to docker.from_env().	None

Example

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().

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

add_image(image, config, base=None)

Adds an image and its build configuration to the orchestrator.

Parameters:

Name	Type	Description	Default
image	Image	The Image instance to add.	required
config	BuildConfig	The build configuration for the image.	required
base	Image	The base image that this image depends on, if any.	None

Returns:

Name	Type	Description
bool		True if the image was added successfully.

Example

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):
    """
    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:
        bool: True if the image was added successfully.

    Example:
        >>> 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()

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

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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"

BuildConfig

Bases: BaseModel

BuildConfig defines the build parameters for constructing a Docker image.

Attributes:

Name	Type	Description
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".

Example

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

Source code in ures/docker/conf.py

class BuildConfig(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".

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

    base_image: str = Field(
        default="python:3.10-slim",
        title="Base Image",
        description="Base image for the container",
    )
    platform: Optional[str] = Field(
        default=None,
        title="Platform",
        description="Platform for the base image in format os[/arch[/variant]]",
    )
    python_deps_manager: Optional[str] = Field(
        default="pip",
        title="Python Dependencies Manager, such as pip or conda",
        description="Define which package manager to use for Python dependencies",
    )
    python_dependencies: Optional[List[str]] = Field(
        default=None,
        title="Python Dependencies",
        description="List of Python dependencies to be installed",
    )
    sys_deps_manager: Optional[str] = Field(
        default="apt",
        title="System Dependencies Manager, such as apt, yum, or apk",
        description="Define which package manager to use for system dependencies",
    )
    sys_dependencies: Optional[List[str]] = Field(
        default=None,
        title="System Dependencies",
        description="List of system dependencies to be installed",
    )
    labels: Optional[List[Tuple[str, str]]] = Field(
        default=None, title="Labels", description="Labels for the image"
    )
    uid: Optional[int] = Field(
        default=None, title="User ID", description="User ID for the container"
    )
    user: Optional[str] = Field(
        default=None, title="User", description="User for the container"
    )
    entrypoint: Optional[List[str]] = Field(
        default=None, title="Entrypoint", description="Entrypoint for the container"
    )
    cmd: Optional[List[str]] = Field(
        default=None, title="Command", description="Command for the container"
    )
    environment: Optional[Dict[str, str]] = Field(
        default=None,
        title="Environment",
        description="Environment variables for the container",
    )
    copies: Optional[List[Dict[str, str]]] = Field(
        default=None, title="Copies", description="Files to be copied to the container"
    )
    context_dir: Union[str, Path] = Field(
        default=Path().cwd(),
        title="Context Directory",
        description="Directory for the build context",
    )
    docker_filename: str = Field(
        default="Dockerfile",
        title="Dockerfile",
        description="Name of the Dockerfile",
    )

    run_commands: Optional[List[str]] = Field(
        default=None,
        title="System Commands",
        description="Commands for running extra commands in the container, suck as installation of nightly pytorch",
    )

    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.

        Returns:
            None

        Example:
            >>> 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))

    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.

        Returns:
            None

        Example:
            >>> 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})

    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.

        Returns:
            None

        Example:
            >>> 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

    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.

        Returns:
            None

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

        Example:
            >>> 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

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

        Returns:
            None

        Example:
            >>> 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)

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

        Returns:
            None

        Example:
            >>> 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)

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

        Returns:
            None

        Example:
            >>> 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)

    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.

        Returns:
            None

        Example:
            >>> 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

    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.

        Returns:
            None

        Example:
            >>> 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

add_copy(src, dest)

Add a file copy instruction to the build configuration.

Parameters:

Name	Type	Description	Default
src	str	Source file path.	required
dest	str	Destination path inside the container.	required

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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(key, value)

Add an environment variable to the build configuration.

Parameters:

Name	Type	Description	Default
key	str	The environment variable name.	required
value	str	The value for the environment variable.	required

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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

add_label(key, value)

Add a label to the build configuration.

Parameters:

Name	Type	Description	Default
key	str	The label key.	required
value	str	The label value.	required

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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_python_dependency(dependency)

Add a Python dependency to be installed in the image.

Parameters:

Name	Type	Description	Default
dependency	str	The Python package dependency (e.g., "flask==2.0.1").	required

Returns:

Type	Description
	None

Example

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

    Returns:
        None

    Example:
        >>> 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_run_command(command)

Add a command to be run during the build process.

Parameters:

Name	Type	Description	Default
command	str	The command to run (e.g., "apt-get update").	required

Returns:

Type	Description
	None

Example

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

    Returns:
        None

    Example:
        >>> 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)

add_system_dependency(dependency)

Add a system dependency to be installed in the image.

Parameters:

Name	Type	Description	Default
dependency	str	The system package dependency (e.g., "curl").	required

Returns:

Type	Description
	None

Example

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

    Returns:
        None

    Example:
        >>> 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)

set_cmd(cmd)

Set the command for the container.

Parameters:

Name	Type	Description	Default
cmd	Union[str, List[str]]	The command(s) to run. If a string is provided, it will be converted to a list.	required

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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

set_context_dir(context_dir)

Set the build context directory.

Parameters:

Name	Type	Description	Default
context_dir	Union[str, Path]	The directory to be used as the build context.	required

Returns:

Type	Description
	None

Raises:

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

Example

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.

    Returns:
        None

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

    Example:
        >>> 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

set_entrypoint(entrypoint)

Set the entrypoint for the container.

Parameters:

Name	Type	Description	Default
entrypoint	Union[str, List[str]]	The entrypoint command(s). If a string is provided, it will be converted to a list.	required

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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

RuntimeConfig

Bases: BaseModel

RuntimeConfig defines the runtime parameters for running a Docker container.

Attributes:

Name	Type	Description
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.

Example

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

Source code in ures/docker/conf.py

class RuntimeConfig(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.

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

    image_name: str = Field(
        default="model-runner",
        title="Image Name",
        description="Name of the image in form of image:tag",
    )
    name: Optional[str] = Field(
        default=None, title="Name", description="Name of the container"
    )
    platform: Optional[str] = Field(
        default=None, title="Platform", description="Platform for the base image"
    )
    detach: bool = Field(
        default=True, title="Detach", description="Run the container in detached mode"
    )
    user: Optional[str] = Field(
        default=None, title="User", description="User for the container"
    )
    remove: bool = Field(
        default=False,
        title="Remove",
        description="Remove the container after it is stopped",
    )
    cpus: Optional[int] = Field(
        default=None,
        title="CPUs",
        description="Number of CPUs to allocate to the container. CPUs allowed (e.g., 0-3, 0,1)",
    )
    gpus: Optional[List[str]] = Field(
        default=None,
        title="GPUs",
        description="List of GPUs to allocate to the container",
    )
    gpu_driver: str = Field(
        default="nvidia",
        title="GPU Driver",
        description="The driver to use for the GPUs, defaults to nvidia",
    )
    memory: Optional[str] = Field(
        default=None,
        title="Memory",
        description="Memory to allocate to the container with a units identification char (100000b, 1000k, 128m, 1g)",
    )
    entrypoint: Optional[List[Union[str, float, int, Path]]] = Field(
        default=None, title="Entrypoint", description="Entrypoint for the container"
    )
    command: Optional[List[Union[int, float, str, Path]]] = Field(
        default=None, title="Command", description="Command for the container"
    )
    env: Optional[Dict[str, str]] = Field(
        default=None,
        title="Environment",
        description="Environment variables for the container",
    )
    volumes: Optional[Dict[str, Dict[str, str]]] = Field(
        default=None, title="Volumes", description="Volumes to mount to the container"
    )
    subnet: Optional[str] = Field(
        default=None, title="Subnet", description="Subnet for the container"
    )
    ipv4: Optional[str] = Field(
        default=None, title="IPv4", description="Specified IPv4 for the container"
    )
    subnet_mask: Optional[str] = Field(
        default="172.17.0.0/16", title="IPv4", description="IPv4 Mask for Subnet"
    )
    subnet_gateway: Optional[str] = Field(
        default="172.17.0.1",
        title="Subnet Gateway",
        description="Subnet Gateway for the container",
    )
    network_mode: Optional[str] = Field(
        default="bridge",
        title="Network Mode",
        description="Network mode for the container",
    )
    out_dir: Path = Field(
        default=Path(get_temp_folder()),
        title="Cache Directory",
        description="Directory for logs, cache, and other temporary files",
    )

    @property
    def log_dir(self) -> Path:
        """
        Get the log directory within the output directory. If it does not exist, it is created.

        Returns:
            Path: The path to the log directory.

        Example:
            >>> config = RuntimeConfig()
            >>> log_directory = config.log_dir
            >>> log_directory.exists()
            True
        """
        log_dir = self.out_dir.joinpath("log")
        if not log_dir.is_dir():
            log_dir.mkdir(parents=True, exist_ok=True)
        return log_dir

    @property
    def cache(self) -> Path:
        """
        Get the cache directory within the output directory. If it does not exist, it is created.

        Returns:
            Path: The path to the cache directory.

        Example:
            >>> config = RuntimeConfig()
            >>> cache_directory = config.cache
            >>> cache_directory.exists()
            True
        """
        cache_dir = self.out_dir.joinpath("cache")
        if not cache_dir.is_dir():
            cache_dir.mkdir(parents=True, exist_ok=True)
        return cache_dir

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

        Returns:
            None

        Example:
            >>> 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}

    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.

        Returns:
            None

        Example:
            >>> 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

cache property

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

Returns:

Name	Type	Description
Path	Path	The path to the cache directory.

Example

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

log_dir property

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

Returns:

Name	Type	Description
Path	Path	The path to the log directory.

Example

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

add_env(key, value)

Add an environment variable for the container.

Parameters:

Name	Type	Description	Default
key	str	The environment variable name.	required
value	str	The value for the environment variable.	required

Returns:

Type	Description
	None

Example

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.

    Returns:
        None

    Example:
        >>> 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

add_volume(host_path, container_path, mode='rw')

Add a volume mapping for the container.

Parameters:

Name	Type	Description	Default
host_path	Union[str, Path]	The path on the host machine.	required
container_path	Union[str, Path]	The destination path inside the container.	required
mode	str	The mode for the volume mapping (e.g., "rw" or "ro"). Defaults to "rw".	'rw'

Returns:

Type	Description
	None

Example

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

    Returns:
        None

    Example:
        >>> 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}