Source code for podman.domain.images

"""Model and Manager for Image resources."""

import logging
from typing import Any, Dict, Iterator, List, Optional, Union

from podman import api
from podman.domain.manager import PodmanResource
from podman.errors import ImageNotFound

logger = logging.getLogger("podman.images")

[docs] class Image(PodmanResource): """Details and configuration for an Image managed by the Podman service.""" def __repr__(self) -> str: return f"""<{self.__class__.__name__}: '{"', '".join(self.tags)}'>""" @property def labels(self): """dict[str, str]: Return labels associated with Image.""" image_labels = self.attrs.get("Labels") if image_labels is None or len(image_labels) == 0: return {} return image_labels @property def tags(self): """list[str]: Return tags from Image.""" repo_tags = self.attrs.get("RepoTags") if repo_tags is None or len(repo_tags) == 0: return [] return [tag for tag in repo_tags if tag != "<none>:<none>"]
[docs] def history(self) -> List[Dict[str, Any]]: """Returns history of the Image. Raises: APIError: when service returns an error """ response = self.client.get(f"/images/{}/history") response.raise_for_status(not_found=ImageNotFound) return response.json()
[docs] def remove( self, **kwargs ) -> List[Dict[api.Literal["Deleted", "Untagged", "Errors", "ExitCode"], Union[str, int]]]: """Delete image from Podman service. Podman only Keyword Args: force: Delete Image even if in use noprune: Ignored. Returns: Report on which images were deleted and untagged, including any reported errors. Raises: ImageNotFound: when image does not exist APIError: when service returns an error """ return self.manager.remove(, **kwargs)
[docs] def save( self, chunk_size: Optional[int] = api.DEFAULT_CHUNK_SIZE, named: Union[str, bool] = False, # pylint: disable=unused-argument ) -> Iterator[bytes]: """Returns Image as tarball. Format is set to docker-archive, this allows load() to import this tarball. Args: chunk_size: If None, data will be streamed in received buffer size. If not None, data will be returned in sized buffers. Default: 2MB named: Ignored. Raises: APIError: when service returns an error """ response = self.client.get( f"/images/{}/get", params={"format": ["docker-archive"]}, stream=True ) response.raise_for_status(not_found=ImageNotFound) return response.iter_content(chunk_size=chunk_size)
[docs] def tag( self, repository: str, tag: Optional[str], force: bool = False, # pylint: disable=unused-argument ) -> bool: """Tag Image into repository. Args: repository: The repository for tagging Image. tag: optional tag name. force: Ignore client errors Returns: True, when operational succeeds. Raises: ImageNotFound: when service cannot find image APIError: when service returns an error """ params = {"repo": repository, "tag": tag} response ="/images/{}/tag", params=params) if response.ok: return True if force and response.status_code <= 500: return False response.raise_for_status(not_found=ImageNotFound) return False