image

ImageConstructor

ImageConstructor(config: BuildConfig)

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.

Initializes the ImageConstructor with the given build configuration.

Parameters:

• config (BuildConfig) –

  The build configuration settings.

Examples:

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

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

home_dir property

home_dir: Path

Returns the home directory path based on the configured user.

Returns:

• Path ( Path ) –

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

Examples:

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

content property

content: List[str]

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

Returns:

• List[str] –

  List[str]: The Dockerfile content lines.

Examples:

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

save

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

Parameters:

• dest (Union[str, Path]) –

  The destination file path or directory.

Returns:

• Path ( Path ) –

  The full path where the Dockerfile was saved.

Examples:

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

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

Image

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

Represents and manages a Docker image.

Attributes:

• _image_name (str) –

  The name of the Docker image.

• _tag (str) –

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

• _client (DockerClient) –

  The Docker client instance.

• _image (Optional[Image]) –

  The Docker image object if available.

Initializes an Image instance.

Parameters:

• image_name (str) –

  The name of the Docker image.

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

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

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

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

Examples:

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

Source code in ures/docker/image.py

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

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

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

name property

name: str

Gets the name of the image.

Returns:

• str ( str ) –

  The image name.

Examples:

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

tag property

tag: str

Gets the tag of the image.

Returns:

• str ( str ) –

  The image tag.

Examples:

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

exist property

exist: bool

Checks if the image exists locally.

Returns:

• bool ( bool ) –

  True if the image exists, False otherwise.

Examples:

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

image property

image: Optional[Image]

Gets the Docker image object.

Returns:

• Optional[Image] –

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

Examples:

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

id property

id: str

Gets the unique ID of the Docker image.

Returns:

• str ( str ) –

  The Docker image ID.

Examples:

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

architecture property

architecture: str

Gets the architecture of the Docker image.

Returns:

• str ( str ) –

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

Examples:

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

image_size property

image_size: int

Gets the size of the Docker image in bytes.

Returns:

• int ( int ) –

  The size of the image in bytes.

Examples:

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

labels property

labels: dict

Gets the labels of the Docker image.

Returns:

• dict ( dict ) –

  A dictionary of image labels.

Examples:

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

get_fullname

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

Constructs the full image name including the tag.

Parameters:

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

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

Returns:

• str ( str ) –

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

Examples:

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

Source code in ures/docker/image.py

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

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

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

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

get_image

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

Retrieves the Docker image from the local repository.

Parameters:

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

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

Returns:

• Optional[Image] –

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

Examples:

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

Source code in ures/docker/image.py

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

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

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

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

pull_image

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

Pulls the Docker image from a remote repository.

Parameters:

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

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

Returns:

• Optional[Image] –

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

Examples:

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

Source code in ures/docker/image.py

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

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

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

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

build_image

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

Builds a Docker image using the specified build configuration.

Parameters:

• build_config (BuildConfig) –

  The configuration for building the image.

• dest (Union[str, Path]) –

  The destination path where the Dockerfile will be saved.

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

  The build context directory. Defaults to build_config.context_dir.

Returns:

• DockerImage ( Image ) –

  The built Docker image.

Examples:

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

Source code in ures/docker/image.py

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

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

    Returns:
        DockerImage: The built Docker image.

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

remove

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

Removes the Docker image from the local repository.

Parameters:

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

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

• force (bool, default: False ) –

  Force removal. Defaults to False.

• noprune (bool, default: False ) –

  Do not remove untagged parent images. Defaults to False.

Examples:

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

Source code in ures/docker/image.py

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

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

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

info

info()

Prints detailed information about the Docker image.

Examples:

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

Source code in ures/docker/image.py

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

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

ImageOrchestrator

ImageOrchestrator(client: Optional[DockerClient] = None)

Orchestrates the building of multiple Docker images considering their dependencies.

Attributes:

• _client (DockerClient) –

  The Docker client instance.

• _images (dict) –

  A dictionary holding images and their build configuration and status.

Initializes the ImageOrchestrator.

Parameters:

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

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

Examples:

>>> orchestrator = ImageOrchestrator()

Source code in ures/docker/image.py

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

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

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

images property

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

Retrieves the dictionary of managed images.

Returns:

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

  A dictionary mapping image fullnames to their configuration and status.

Examples:

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

add_image

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

Adds an image and its build configuration to the orchestrator.

Parameters:

• image (Image) –

  The Image instance to add.

• config (BuildConfig) –

  The build configuration for the image.

• base (Image, default: None ) –

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

Returns:

• bool –

  True if the image was registered in the orchestrator.

Examples:

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

Source code in ures/docker/image.py

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

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

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

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

build_all

build_all()

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

Examples:

>>> orchestrator.build_all()

Source code in ures/docker/image.py

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

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

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