Index

wrknv.container.runtime

TODO: Add module docstring.

Classes

ContainerRuntime

Bases: ABC

Abstract base class for container runtimes (Docker, Podman, etc.).

Functions

build_image abstractmethod

build_image(
    dockerfile: str,
    tag: str,
    context: str = ".",
    build_args: dict[str, str] | None = None,
    **extra_options: Any
) -> CompletedProcess

Build a container image.

Parameters:

Name	Type	Description	Default
dockerfile	str	Path to Dockerfile	required
tag	str	Image tag	required
context	str	Build context directory	'.'
build_args	dict[str, str] | None	Build arguments	None
**extra_options	Any	Runtime-specific options	{}

Returns:

Type	Description
CompletedProcess	CompletedProcess

Source code in wrknv/container/runtime/base.py

@abstractmethod
def build_image(
    self,
    dockerfile: str,
    tag: str,
    context: str = ".",
    build_args: dict[str, str] | None = None,
    **extra_options: Any,
) -> CompletedProcess:
    """Build a container image.

    Args:
        dockerfile: Path to Dockerfile
        tag: Image tag
        context: Build context directory
        build_args: Build arguments
        **extra_options: Runtime-specific options

    Returns:
        CompletedProcess
    """
    pass

container_exists abstractmethod

container_exists(name: str) -> bool

Check if container exists.

Parameters:

Name	Type	Description	Default
name	str	Container name	required

Returns:

Type	Description
bool	True if container exists

Source code in wrknv/container/runtime/base.py

@abstractmethod
def container_exists(self, name: str) -> bool:
    """Check if container exists.

    Args:
        name: Container name

    Returns:
        True if container exists
    """
    pass

container_running abstractmethod

container_running(name: str) -> bool

Check if container is running.

Parameters:

Name	Type	Description	Default
name	str	Container name	required

Returns:

Type	Description
bool	True if container is running

Source code in wrknv/container/runtime/base.py

@abstractmethod
def container_running(self, name: str) -> bool:
    """Check if container is running.

    Args:
        name: Container name

    Returns:
        True if container is running
    """
    pass

exec_in_container abstractmethod

exec_in_container(
    name: str,
    command: list[str],
    interactive: bool = False,
    tty: bool = False,
    user: str | None = None,
    workdir: str | None = None,
    environment: dict[str, str] | None = None,
) -> CompletedProcess

Execute command in a running container.

Parameters:

Name	Type	Description	Default
name	str	Container name	required
command	list[str]	Command to execute	required
interactive	bool	Keep STDIN open	False
tty	bool	Allocate pseudo-TTY	False
user	str | None	User to run as	None
workdir	str | None	Working directory	None
environment	dict[str, str] | None	Environment variables	None

Returns:

Type	Description
CompletedProcess	CompletedProcess with command output

Source code in wrknv/container/runtime/base.py

@abstractmethod
def exec_in_container(
    self,
    name: str,
    command: list[str],
    interactive: bool = False,
    tty: bool = False,
    user: str | None = None,
    workdir: str | None = None,
    environment: dict[str, str] | None = None,
) -> CompletedProcess:
    """Execute command in a running container.

    Args:
        name: Container name
        command: Command to execute
        interactive: Keep STDIN open
        tty: Allocate pseudo-TTY
        user: User to run as
        workdir: Working directory
        environment: Environment variables

    Returns:
        CompletedProcess with command output
    """
    pass

get_container_logs abstractmethod

get_container_logs(
    name: str,
    follow: bool = False,
    tail: int | None = None,
    since: str | None = None,
) -> CompletedProcess

Get container logs.

Parameters:

Name	Type	Description	Default
name	str	Container name	required
follow	bool	Follow log output	False
tail	int | None	Number of lines to tail	None
since	str | None	Show logs since timestamp	None

Returns:

Type	Description
CompletedProcess	CompletedProcess with logs in stdout

Source code in wrknv/container/runtime/base.py

@abstractmethod
def get_container_logs(
    self,
    name: str,
    follow: bool = False,
    tail: int | None = None,
    since: str | None = None,
) -> CompletedProcess:
    """Get container logs.

    Args:
        name: Container name
        follow: Follow log output
        tail: Number of lines to tail
        since: Show logs since timestamp

    Returns:
        CompletedProcess with logs in stdout
    """
    pass

inspect_container abstractmethod

inspect_container(name: str) -> dict[str, Any]

Get detailed container information.

Parameters:

Name	Type	Description	Default
name	str	Container name	required

Returns:

Type	Description
dict[str, Any]	Container inspection data

Source code in wrknv/container/runtime/base.py

@abstractmethod
def inspect_container(self, name: str) -> dict[str, Any]:
    """Get detailed container information.

    Args:
        name: Container name

    Returns:
        Container inspection data
    """
    pass

is_available abstractmethod

is_available() -> bool

Check if the container runtime is available.

Returns:

Type	Description
bool	True if runtime is available and functional

Source code in wrknv/container/runtime/base.py

@abstractmethod
def is_available(self) -> bool:
    """Check if the container runtime is available.

    Returns:
        True if runtime is available and functional
    """
    pass

list_containers abstractmethod

list_containers(all: bool = False) -> list[dict[str, Any]]

List containers.

Parameters:

Name	Type	Description	Default
all	bool	Include stopped containers	False

Returns:

Type	Description
list[dict[str, Any]]	List of container information dicts

Source code in wrknv/container/runtime/base.py

@abstractmethod
def list_containers(self, all: bool = False) -> list[dict[str, Any]]:
    """List containers.

    Args:
        all: Include stopped containers

    Returns:
        List of container information dicts
    """
    pass

remove_container abstractmethod

remove_container(
    name: str, force: bool = False
) -> CompletedProcess

Remove a container.

Parameters:

Name	Type	Description	Default
name	str	Container name	required
force	bool	Force removal of running container	False

Returns:

Type	Description
CompletedProcess	CompletedProcess

Source code in wrknv/container/runtime/base.py

@abstractmethod
def remove_container(self, name: str, force: bool = False) -> CompletedProcess:
    """Remove a container.

    Args:
        name: Container name
        force: Force removal of running container

    Returns:
        CompletedProcess
    """
    pass

run_container abstractmethod

run_container(
    image: str,
    name: str,
    detach: bool = True,
    volumes: list[str] | None = None,
    environment: dict[str, str] | None = None,
    ports: list[str] | None = None,
    workdir: str | None = None,
    command: list[str] | None = None,
    **extra_options: Any
) -> CompletedProcess

Start a new container.

Parameters:

Name	Type	Description	Default
image	str	Container image to run	required
name	str	Container name	required
detach	bool	Run in detached mode	True
volumes	list[str] | None	Volume mappings (host:container)	None
environment	dict[str, str] | None	Environment variables	None
ports	list[str] | None	Port mappings (host:container)	None
workdir	str | None	Working directory inside container	None
command	list[str] | None	Command to run in container	None
**extra_options	Any	Runtime-specific options	{}

Returns:

Type	Description
CompletedProcess	CompletedProcess with container ID in stdout

Source code in wrknv/container/runtime/base.py

@abstractmethod
def run_container(
    self,
    image: str,
    name: str,
    detach: bool = True,
    volumes: list[str] | None = None,
    environment: dict[str, str] | None = None,
    ports: list[str] | None = None,
    workdir: str | None = None,
    command: list[str] | None = None,
    **extra_options: Any,
) -> CompletedProcess:
    """Start a new container.

    Args:
        image: Container image to run
        name: Container name
        detach: Run in detached mode
        volumes: Volume mappings (host:container)
        environment: Environment variables
        ports: Port mappings (host:container)
        workdir: Working directory inside container
        command: Command to run in container
        **extra_options: Runtime-specific options

    Returns:
        CompletedProcess with container ID in stdout
    """
    pass

start_container abstractmethod

start_container(name: str) -> CompletedProcess

Start an existing container.

Parameters:

Name	Type	Description	Default
name	str	Container name	required

Returns:

Type	Description
CompletedProcess	CompletedProcess

Source code in wrknv/container/runtime/base.py

@abstractmethod
def start_container(self, name: str) -> CompletedProcess:
    """Start an existing container.

    Args:
        name: Container name

    Returns:
        CompletedProcess
    """
    pass

stop_container abstractmethod

stop_container(
    name: str, timeout: int = 10
) -> CompletedProcess

Stop a running container.

Parameters:

Name	Type	Description	Default
name	str	Container name	required
timeout	int	Seconds to wait before force stopping	10

Returns:

Type	Description
CompletedProcess	CompletedProcess

Source code in wrknv/container/runtime/base.py

@abstractmethod
def stop_container(self, name: str, timeout: int = 10) -> CompletedProcess:
    """Stop a running container.

    Args:
        name: Container name
        timeout: Seconds to wait before force stopping

    Returns:
        CompletedProcess
    """
    pass

DockerRuntime

Bases: ContainerRuntime

Docker implementation of container runtime.

Functions

build_image

build_image(
    dockerfile: str,
    tag: str,
    context: str,
    build_args: dict[str, str] | None,
    **extra_options: Any
) -> CompletedProcess

Build a Docker image.

Source code in wrknv/container/runtime/docker.py

def build_image(
    self,
    dockerfile: str,
    tag: str,
    context: str,
    build_args: dict[str, str] | None,
    **extra_options: Any,
) -> CompletedProcess:
    """Build a Docker image."""
    cmd = [self.runtime_command, "build", "-f", dockerfile, "-t", tag]

    for key, value in (build_args or {}).items():
        cmd.extend(["--build-arg", f"{key}={value}"])

    if extra_options.get("no_cache"):
        cmd.append("--no-cache")
    if extra_options.get("platform"):
        cmd.extend(["--platform", extra_options["platform"]])

    cmd.append(context)

    try:
        result = run(cmd, check=True)
        logger.info("Docker image built", tag=tag, dockerfile=dockerfile)
        return result
    except ProcessError as e:
        logger.error(
            "Docker build failed",
            tag=tag,
            dockerfile=dockerfile,
            error=str(e),
        )
        raise

container_exists

container_exists(name: str) -> bool

Check if Docker container exists.

Source code in wrknv/container/runtime/docker.py

def container_exists(self, name: str) -> bool:
    """Check if Docker container exists."""
    try:
        result = run([self.runtime_command, "ps", "-a", "--format", "{{.Names}}"], check=False)
        return name in result.stdout.splitlines() if result.stdout else False
    except ProcessError:
        return False

container_running

container_running(name: str) -> bool

Check if Docker container is running.

Source code in wrknv/container/runtime/docker.py

def container_running(self, name: str) -> bool:
    """Check if Docker container is running."""
    try:
        result = run([self.runtime_command, "ps", "--format", "{{.Names}}"], check=False)
        return name in result.stdout.splitlines() if result.stdout else False
    except ProcessError:
        return False

exec_in_container

exec_in_container(
    name: str,
    command: list[str],
    interactive: bool,
    tty: bool,
    user: str | None,
    workdir: str | None,
    environment: dict[str, str] | None,
) -> CompletedProcess

Execute command in a running Docker container.

Source code in wrknv/container/runtime/docker.py

def exec_in_container(
    self,
    name: str,
    command: list[str],
    interactive: bool,
    tty: bool,
    user: str | None,
    workdir: str | None,
    environment: dict[str, str] | None,
) -> CompletedProcess:
    """Execute command in a running Docker container."""
    cmd = [self.runtime_command, "exec"]

    if interactive:
        cmd.append("-i")
    if tty:
        cmd.append("-t")
    if user:
        cmd.extend(["-u", user])
    if workdir:
        cmd.extend(["-w", workdir])

    for key, value in (environment or {}).items():
        cmd.extend(["-e", f"{key}={value}"])

    cmd.append(name)
    cmd.extend(command)

    try:
        # Note: Interactive mode may need special handling
        # Check if foundation.process supports it
        result = run(cmd, check=True)
        logger.debug(
            "Docker exec completed",
            name=name,
            command=command,
            interactive=interactive,
        )
        return result
    except ProcessError as e:
        logger.error(
            "Docker exec failed",
            name=name,
            command=command,
            error=str(e),
        )
        raise

get_container_logs

get_container_logs(
    name: str,
    follow: bool,
    tail: int | None,
    since: str | None,
) -> CompletedProcess

Get Docker container logs.

Source code in wrknv/container/runtime/docker.py

def get_container_logs(
    self,
    name: str,
    follow: bool,
    tail: int | None,
    since: str | None,
) -> CompletedProcess:
    """Get Docker container logs."""
    cmd = [self.runtime_command, "logs"]

    if follow:
        cmd.append("-f")
    if tail is not None:
        cmd.extend(["--tail", str(tail)])
    if since:
        cmd.extend(["--since", since])

    cmd.append(name)

    try:
        result = run(cmd, check=True)
        return result
    except ProcessError as e:
        logger.error("Failed to get Docker logs", name=name, error=str(e))
        raise

inspect_container

inspect_container(name: str) -> dict[str, Any]

Get detailed Docker container information.

Source code in wrknv/container/runtime/docker.py

def inspect_container(self, name: str) -> dict[str, Any]:
    """Get detailed Docker container information."""
    try:
        result = run([self.runtime_command, "inspect", name], check=True)
        if result.stdout:
            data = json.loads(result.stdout)
            return data[0] if data else {}
        return {}
    except (ProcessError, json.JSONDecodeError) as e:
        logger.error("Failed to inspect Docker container", name=name, error=str(e))
        return {}

is_available

is_available() -> bool

Check if Docker is available.

Uses circuit breaker to prevent repeated checks when Docker is unavailable. If circuit is open, raises RuntimeError which callers should catch.

Source code in wrknv/container/runtime/docker.py

@circuit_breaker(
    failure_threshold=3,
    recovery_timeout=30.0,
    expected_exception=(ProcessError, OSError),
)
def is_available(self) -> bool:
    """Check if Docker is available.

    Uses circuit breaker to prevent repeated checks when Docker is unavailable.
    If circuit is open, raises RuntimeError which callers should catch.
    """
    try:
        result = run([self.runtime_command, "version"], check=False)
        if result.returncode != 0:
            # Treat non-zero exit as failure to trigger circuit breaker
            raise ProcessError(
                message="Docker not available",
                command=[self.runtime_command, "version"],
                returncode=result.returncode,
                stdout=result.stdout,
                stderr=result.stderr,
            )
        return True
    except (ProcessError, OSError):
        # Re-raise to let circuit breaker count this as a failure
        raise

list_containers

list_containers(all: bool) -> list[dict[str, Any]]

List Docker containers.

Source code in wrknv/container/runtime/docker.py

def list_containers(self, all: bool) -> list[dict[str, Any]]:
    """List Docker containers."""
    cmd = [self.runtime_command, "ps", "--format", "json"]
    if all:
        cmd.append("-a")

    try:
        result = run(cmd, check=True)
        containers = []
        if result.stdout:
            for line in result.stdout.strip().splitlines():
                if line:
                    containers.append(json.loads(line))
        return containers
    except (ProcessError, json.JSONDecodeError) as e:
        logger.error("Failed to list Docker containers", error=str(e))
        return []

remove_container

remove_container(
    name: str, force: bool
) -> CompletedProcess

Remove a Docker container.

Source code in wrknv/container/runtime/docker.py

def remove_container(self, name: str, force: bool) -> CompletedProcess:
    """Remove a Docker container."""
    cmd = [self.runtime_command, "rm"]
    if force:
        cmd.append("-f")
    cmd.append(name)

    try:
        result = run(cmd, check=True)
        logger.info("Docker container removed", name=name, forced=force)
        return result
    except ProcessError as e:
        logger.error("Failed to remove Docker container", name=name, error=str(e))
        raise

run_container

run_container(
    image: str,
    name: str,
    detach: bool,
    volumes: list[str] | None,
    environment: dict[str, str] | None,
    ports: list[str] | None,
    workdir: str | None,
    command: list[str] | None,
    **extra_options: Any
) -> CompletedProcess

Start a new Docker container.

Source code in wrknv/container/runtime/docker.py

def run_container(
    self,
    image: str,
    name: str,
    detach: bool,
    volumes: list[str] | None,
    environment: dict[str, str] | None,
    ports: list[str] | None,
    workdir: str | None,
    command: list[str] | None,
    **extra_options: Any,
) -> CompletedProcess:
    """Start a new Docker container."""
    cmd = [self.runtime_command, "run"]

    if detach:
        cmd.append("-d")

    cmd.extend(["--name", name])

    # Add volumes
    for volume in volumes or []:
        cmd.extend(["-v", volume])

    # Add environment variables
    for key, value in (environment or {}).items():
        cmd.extend(["-e", f"{key}={value}"])

    # Add ports
    for port in ports or []:
        cmd.extend(["-p", port])

    # Add workdir
    if workdir:
        cmd.extend(["--workdir", workdir])

    # Add extra Docker-specific options
    if extra_options.get("restart"):
        cmd.extend(["--restart", extra_options["restart"]])
    if extra_options.get("network"):
        cmd.extend(["--network", extra_options["network"]])
    if extra_options.get("hostname"):
        cmd.extend(["--hostname", extra_options["hostname"]])

    # Add image
    cmd.append(image)

    # Add command
    if command:
        cmd.extend(command)

    try:
        result = run(cmd, check=True)
        logger.info(
            "Docker container started",
            name=name,
            image=image,
            container_id=result.stdout.strip()[:12] if result.stdout else None,
        )
        return result
    except ProcessError as e:
        logger.error(
            "Failed to start Docker container",
            name=name,
            image=image,
            error=str(e),
            stderr=e.stderr,
        )
        raise

start_container

start_container(name: str) -> CompletedProcess

Start an existing Docker container.

Source code in wrknv/container/runtime/docker.py

def start_container(self, name: str) -> CompletedProcess:
    """Start an existing Docker container."""
    try:
        result = run([self.runtime_command, "start", name], check=True)
        logger.info("Docker container started", name=name)
        return result
    except ProcessError as e:
        logger.error("Failed to start Docker container", name=name, error=str(e))
        raise

stop_container

stop_container(name: str, timeout: int) -> CompletedProcess

Stop a running Docker container.

Source code in wrknv/container/runtime/docker.py

def stop_container(self, name: str, timeout: int) -> CompletedProcess:
    """Stop a running Docker container."""
    try:
        result = run([self.runtime_command, "stop", "-t", str(timeout), name], check=True)
        logger.info("Docker container stopped", name=name)
        return result
    except ProcessError as e:
        logger.error("Failed to stop Docker container", name=name, error=str(e))
        raise