napistu.utils.docker_utils

Base Docker image management for Napistu workflows.

Provides image validation, availability checking, and container execution.

Classes

ImageInfo: Docker image information.

Classes

`DockerImageManager`(image_info[, auto_pull])	Base class for managing Docker images.
`ImageInfo`(*, name, tag, registry[, platform])	Docker image information with registry configuration.

class napistu.utils.docker_utils.DockerImageManager(image_info: ImageInfo, auto_pull: bool = False)

Bases: ABC

Base class for managing Docker images.

Provides image validation, availability checking, and container execution. Subclasses implement specific image requirements and entrypoint access.

image_info

Image specification including registry

Type:: ImageInfo

auto_pull

Automatically pull image if not available locally (ignored for local images)

Type:: bool

Properties

----------

full_image_name

Get full image name with tag

Type:: str

Public Methods

--------------

ensure_available: Ensure image is available, pulling if necessary

is_available: Check if Docker image exists locally

pull: Pull image from registry

run_command(cmd

Run a command in the Docker container

Type:: List[str], volumes: Optional[Dict[str, str]] = None, environment: Optional[Dict[str, str]] = None, **kwargs) -> subprocess.CompletedProcess

Examples

Remote image from Docker Hub: >>> info = ImageInfo(name=”username/diffdock”, tag=”latest”) >>> manager = SomeDockerImage(info, auto_pull=True)

Local-only image: >>> info = ImageInfo(name=”diffdock-arm”, tag=”latest”, registry=”local”) >>> manager = DiffDockImageARM(info)

GCR image: >>> info = ImageInfo(name=”project/diffdock”, tag=”v1.0”, registry=”gcr.io”) >>> manager = SomeDockerImage(info, auto_pull=True)

__init__(image_info: ImageInfo, auto_pull: bool = False)

ensure_available() → bool

Ensure image is available, pulling if necessary.

Returns:: True if image is available (or was successfully pulled)
Return type:: bool
Raises:: RuntimeError – If image cannot be made available

is_available() → bool

Check if Docker image exists locally.

For local images, checks the local name. For remote images, checks the full registry path.

Returns:: True if image is available locally
Return type:: bool

pull() → bool

Pull image from registry.

Returns:: True if pull succeeded, False if failed or local-only image
Return type:: bool

run_command(cmd: List[str], volumes: Dict[str, str] | None = None, environment: Dict[str, str] | None = None, **kwargs) → CompletedProcess

Run a command in the Docker container.

Parameters:

cmd (List[str]) – Command to run inside container
volumes (Dict[str, str], optional) – Volume mounts {host_path: container_path} or {host_path: “container_path:mode”} Mode can be ‘ro’ (read-only) or ‘rw’ (read-write, default)
environment (Dict[str, str], optional) – Environment variables
**kwargs – Additional arguments passed to subprocess.run

Returns:

Result of the command execution

Return type:

subprocess.CompletedProcess

_abc_impl = <_abc._abc_data object>

property full_image_name: str: Get full image name with tag.

class napistu.utils.docker_utils.ImageInfo(*, name: str, tag: str, registry: str, platform: str | None = None)

Bases: BaseModel

Docker image information with registry configuration.

name

Image name without registry prefix (e.g., “project/repo/image”)

Type:: str

tag

Image tag

Type:: str

registry

Registry URL. Valid values: - “local”: Pre-built local image, no pulling - “docker.io”: Docker Hub - “ghcr.io”: GitHub Container Registry - Regional Artifact Registry: “us-west1-docker.pkg.dev”, “us-central1-docker.pkg.dev”, etc. - Custom registry URLs

Type:: str, optional

platform

Target platform (e.g., “linux/arm64”, “linux/amd64”)

Type:: Optional[str]

Properties

----------

is_local

Check if this is a local-only image

Type:: bool

full_name

Get full image name with registry and tag

Type:: str

pull_name

Get the name to use for pulling

Type:: Optional[str]

Examples

Local-only image: >>> info = ImageInfo(name=”diffdock-arm”, tag=”latest”, registry=”local”) >>> info.full_name ‘diffdock-arm:latest’

Google Artifact Registry (public): >>> info = ImageInfo( … name=”shackett/napistu-images/napistu-base”, … tag=”latest”, … registry=”us-west1-docker.pkg.dev” … ) >>> info.full_name ‘us-west1-docker.pkg.dev/shackett/napistu-images/napistu-base:latest’

_abc_impl = <_abc._abc_data object>

property full_name: str

Get full image name with registry and tag.

Returns:: Full image reference: - Local: “name:tag” - Remote: “registry/name:tag”
Return type:: str

property is_local: bool: Check if this is a local-only image.

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

platform: str | None

property pull_name: str | None

Get the name to use for pulling.

Returns:: Full pull reference, or None if local-only
Return type:: str or None

registry: str

tag: str

napistu.utils.docker_utils._verify_docker_available() → None

Verify Docker is installed and running.

Raises:: RuntimeError – If Docker is not available or not running