Index

wrknv.container.operations

TODO: Add module docstring.

Classes

ContainerBuilder

Handles container build operations.

Functions

build

build(
    dockerfile: str,
    tag: str,
    context: str,
    build_args: dict[str, str] | None,
    stream_output: bool,
    **extra_options: Any
) -> bool

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	required
build_args	dict[str, str] | None	Build arguments	required
stream_output	bool	Whether to stream build output	required
**extra_options	Any	Runtime-specific build options	{}

Returns:

Type	Description
bool	True if build successful

Source code in wrknv/container/operations/build.py

def build(
    self,
    dockerfile: str,
    tag: str,
    context: str,
    build_args: dict[str, str] | None,
    stream_output: bool,
    **extra_options: Any,
) -> bool:
    """Build a container image.

    Args:
        dockerfile: Path to Dockerfile
        tag: Image tag
        context: Build context directory
        build_args: Build arguments
        stream_output: Whether to stream build output
        **extra_options: Runtime-specific build options

    Returns:
        True if build successful
    """
    try:
        if stream_output:
            # Build command for streaming
            cmd = self._build_command(
                dockerfile=dockerfile, tag=tag, context=context, build_args=build_args, **extra_options
            )

            self.console.print(f"[cyan]🔨 Building image {tag}...[/cyan]")

            # Stream build output
            for line in stream(cmd):
                self.console.print(line, end="")

            logger.info("Image built successfully", tag=tag, dockerfile=dockerfile)
            return True
        else:
            # Use runtime's build method
            self.runtime.build_image(
                dockerfile=dockerfile, tag=tag, context=context, build_args=build_args, **extra_options
            )

            return True

    except ProcessError as e:
        logger.error(
            "Build failed",
            tag=tag,
            dockerfile=dockerfile,
            error=str(e),
            stderr=e.stderr,
        )
        self.console.print(f"[red]❌ Build failed: {e}[/red]")
        if e.stderr:
            self.console.print(f"[red]{e.stderr}[/red]")
        return False

generate_dockerfile

generate_dockerfile(
    container_config: ContainerConfig,
) -> str

Generate Dockerfile content from configuration.

Parameters:

Name	Type	Description	Default
container_config	ContainerConfig	Container configuration	required

Returns:

Type	Description
str	Dockerfile content as string

Source code in wrknv/container/operations/build.py

def generate_dockerfile(self, container_config: ContainerConfig) -> str:
    """
    Generate Dockerfile content from configuration.

    Args:
        container_config: Container configuration

    Returns:
        Dockerfile content as string
    """
    # Use configured base image or default
    base_image = container_config.base_image or "ubuntu:22.04"

    # Start with base image
    lines = [f"FROM {base_image}", ""]

    # Set working directory
    lines.extend(["WORKDIR /workspace", ""])

    # Install system packages
    if container_config.additional_packages:
        packages = " ".join(container_config.additional_packages)
        lines.extend(
            [
                "RUN apt-get update && apt-get install -y \\",
                f"    {packages} \\",
                "    && rm -rf /var/lib/apt/lists/*",
                "",
            ]
        )
    else:
        # Install default packages
        lines.extend(
            [
                "RUN apt-get update && apt-get install -y \\",
                "    curl \\",
                "    git \\",
                "    && rm -rf /var/lib/apt/lists/*",
                "",
            ]
        )

    # Install Python if python_version is specified and base image isn't already Python
    if container_config.python_version and not base_image.startswith("python:"):
        py_version = container_config.python_version
        lines.extend(
            [
                f"RUN apt-get update && apt-get install -y python{py_version} python{py_version}-venv \\",
                "    && rm -rf /var/lib/apt/lists/*",
                "",
            ]
        )

    # Set environment variables
    if container_config.environment:
        for key, value in container_config.environment.items():
            lines.append(f"ENV {key}={value}")
        lines.append("")

    # Create user and set ownership
    lines.extend(
        [
            "RUN useradd -m -s /bin/bash user",
            "RUN chown -R user:user /workspace",
            "USER user",
            "",
        ]
    )

    # Keep container running
    lines.extend(
        [
            "# Keep container running",
            'CMD ["sleep", "infinity"]',
        ]
    )

    return "\n".join(lines)

image_exists

image_exists(tag: str) -> bool

Check if an image exists locally.

Parameters:

Name	Type	Description	Default
tag	str	Image tag to check	required

Returns:

Type	Description
bool	True if image exists

Source code in wrknv/container/operations/build.py

def image_exists(self, tag: str) -> bool:
    """Check if an image exists locally.

    Args:
        tag: Image tag to check

    Returns:
        True if image exists
    """
    try:
        from provide.foundation.process import run

        result = run(
            [self.runtime.runtime_command, "images", "--format", "{{.Repository}}:{{.Tag}}"],
            check=False,
        )

        if result.stdout:
            images = result.stdout.strip().splitlines()
            return tag in images

        return False

    except ProcessError:
        return False

push_image

push_image(tag: str) -> bool

Push image to registry.

Parameters:

Name	Type	Description	Default
tag	str	Image tag to push	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/build.py

def push_image(self, tag: str) -> bool:
    """Push image to registry.

    Args:
        tag: Image tag to push

    Returns:
        True if successful
    """
    try:
        from provide.foundation.process import run

        self.console.print(f"[cyan]📤 Pushing image {tag}...[/cyan]")

        run([self.runtime.runtime_command, "push", tag], check=True)

        logger.info("Image pushed", tag=tag)
        return True

    except ProcessError as e:
        logger.error("Failed to push image", tag=tag, error=str(e))
        self.console.print(f"[red]❌ Failed to push image: {e}[/red]")
        return False

tag_image

tag_image(source_tag: str, target_tag: str) -> bool

Tag an existing image.

Parameters:

Name	Type	Description	Default
source_tag	str	Source image tag	required
target_tag	str	Target image tag	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/build.py

def tag_image(self, source_tag: str, target_tag: str) -> bool:
    """Tag an existing image.

    Args:
        source_tag: Source image tag
        target_tag: Target image tag

    Returns:
        True if successful
    """
    try:
        from provide.foundation.process import run

        run([self.runtime.runtime_command, "tag", source_tag, target_tag], check=True)

        logger.info("Image tagged", source=source_tag, target=target_tag)
        return True

    except ProcessError as e:
        logger.error(
            "Failed to tag image",
            source=source_tag,
            target=target_tag,
            error=str(e),
        )
        self.console.print(f"[red]❌ Failed to tag image: {e}[/red]")
        return False

ContainerExec

Handles container exec operations.

Functions

enter

enter(shell: str | None, **kwargs: Any) -> bool

Enter the container with an interactive shell.

Parameters:

Name	Type	Description	Default
shell	str | None	Shell to use	required
**kwargs	Any	Additional exec options	{}

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/exec.py

def enter(self, shell: str | None, **kwargs: Any) -> bool:
    """Enter the container with an interactive shell.

    Args:
        shell: Shell to use
        **kwargs: Additional exec options

    Returns:
        True if successful
    """
    return self.exec(command=None, shell=shell, interactive=True, tty=True, **kwargs)

exec

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

Execute a command in the container.

Parameters:

Name	Type	Description	Default
command	list[str] | None	Command to execute (defaults to shell)	None
shell	str | None	Shell to use (defaults to /bin/bash or /bin/sh)	None
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
bool	True if successful

Source code in wrknv/container/operations/exec.py

@resilient
def exec(
    self,
    command: list[str] | None = None,
    shell: str | None = None,
    interactive: bool = False,
    tty: bool = False,
    user: str | None = None,
    workdir: str | None = None,
    environment: dict[str, str] | None = None,
) -> bool:
    """Execute a command in the container.

    Args:
        command: Command to execute (defaults to shell)
        shell: Shell to use (defaults to /bin/bash or /bin/sh)
        interactive: Keep STDIN open
        tty: Allocate pseudo-TTY
        user: User to run as
        workdir: Working directory
        environment: Environment variables

    Returns:
        True if successful
    """
    try:
        # Check if container is running
        if not self.runtime.container_running(self.container_name):
            self.console.print(f"[red]❌ Container {self.container_name} is not running[/red]")
            return False

        # Determine command to run
        if command is None:
            # Use shell
            if shell is None:
                # Try to detect available shell
                shell = self._detect_shell()
            command = [shell]

        # Execute command

        # For interactive commands, we need to use os.system or similar
        # as foundation.process might not support interactive TTY yet
        if interactive and tty:
            # Build command for os.system
            cmd_str = self._build_exec_command(
                command=command,
                interactive=True,
                tty=True,
                user=user,
                workdir=workdir,
                environment=environment,
            )

            # Use os.system for interactive TTY support
            result = os.system(cmd_str)
            return result == 0

        else:
            # Non-interactive, use foundation.process
            result = self.runtime.exec_in_container(
                name=self.container_name,
                command=command,
                interactive=interactive,
                tty=tty,
                user=user,
                workdir=workdir,
                environment=environment,
            )

            if result.stdout:
                self.console.print(result.stdout)

            return True

    except ProcessError as e:
        logger.error(
            "Container exec failed",
            container=self.container_name,
            command=command,
            error=str(e),
            stderr=e.stderr,
        )
        self.console.print(f"[red]❌ Exec failed: {e}[/red]")
        return False

run_command

run_command(
    command: list[str], capture_output: bool, **kwargs: Any
) -> str | None

Run a command in the container and return output.

Parameters:

Name	Type	Description	Default
command	list[str]	Command to run	required
capture_output	bool	Whether to capture output	required
**kwargs	Any	Additional exec options	{}

Returns:

Type	Description
str | None	Command output if capture_output is True, else None

Source code in wrknv/container/operations/exec.py

@resilient
def run_command(
    self,
    command: list[str],
    capture_output: bool,
    **kwargs: Any,
) -> str | None:
    """Run a command in the container and return output.

    Args:
        command: Command to run
        capture_output: Whether to capture output
        **kwargs: Additional exec options

    Returns:
        Command output if capture_output is True, else None
    """
    try:
        result = self.runtime.exec_in_container(
            name=self.container_name,
            command=command,
            interactive=False,
            tty=False,
            user=kwargs.get("user"),
            workdir=kwargs.get("workdir"),
            environment=kwargs.get("environment"),
        )

        if capture_output:
            return result.stdout

        if result.stdout:
            self.console.print(result.stdout)

        return None

    except ProcessError as e:
        logger.error(
            "Command execution failed",
            container=self.container_name,
            command=command,
            error=str(e),
        )
        if capture_output:
            return None
        self.console.print(f"[red]❌ Command failed: {e}[/red]")
        return None

ContainerLifecycle

Manages container lifecycle operations.

Functions

exists

exists() -> bool

Check if container exists.

Source code in wrknv/container/operations/lifecycle.py

def exists(self) -> bool:
    """Check if container exists."""
    return self.runtime.container_exists(self.container_name)

is_running

is_running() -> bool

Check if container is running.

Source code in wrknv/container/operations/lifecycle.py

def is_running(self) -> bool:
    """Check if container is running."""
    return self.runtime.container_running(self.container_name)

remove

remove(force: bool) -> bool

Remove the container.

Parameters:

Name	Type	Description	Default
force	bool	Force removal even if running	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/lifecycle.py

def remove(self, force: bool) -> bool:
    """Remove the container.

    Args:
        force: Force removal even if running

    Returns:
        True if successful
    """
    try:
        if not self.runtime.container_exists(self.container_name):
            self.console.print(f"[yellow]⚠️  Container {self.container_name} does not exist[/yellow]")
            return True

        # Stop first if running and not forcing
        if not force and self.runtime.container_running(self.container_name):
            self.console.print(
                f"[yellow]⚠️  Stopping container {self.container_name} before removal[/yellow]"
            )
            if not self.stop():
                return False

        self.console.print(f"🗑️  Removing container {self.container_name}...")

        self.runtime.remove_container(self.container_name, force=force)

        return True

    except ProcessError as e:
        logger.error(
            "Failed to remove container",
            name=self.container_name,
            error=str(e),
        )
        self.console.print(f"[red]❌ Failed to remove container: {e}[/red]")
        return False

restart

restart(timeout: int) -> bool

Restart the container.

Parameters:

Name	Type	Description	Default
timeout	int	Seconds to wait before force stopping	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/lifecycle.py

def restart(self, timeout: int) -> bool:
    """Restart the container.

    Args:
        timeout: Seconds to wait before force stopping

    Returns:
        True if successful
    """
    self.console.print(f"{self.restart_emoji} Restarting container {self.container_name}...")

    # Stop if running
    if self.runtime.container_running(self.container_name) and not self.stop(timeout=timeout):
        return False

    # Start again
    return self.start(create_if_missing=False)

start

start(create_if_missing: bool, **run_options: Any) -> bool

Start the container.

Parameters:

Name	Type	Description	Default
create_if_missing	bool	Create container if it doesn't exist	required
**run_options	Any	Options for container creation	{}

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/lifecycle.py

def start(self, create_if_missing: bool, **run_options: Any) -> bool:
    """Start the container.

    Args:
        create_if_missing: Create container if it doesn't exist
        **run_options: Options for container creation

    Returns:
        True if successful
    """
    try:
        # Check if container exists
        if self.runtime.container_exists(self.container_name):
            # Container exists, just start it
            if self.runtime.container_running(self.container_name):
                self.console.print(
                    f"[yellow]⚠️  Container {self.container_name} is already running[/yellow]"
                )
                return True

            self.console.print(f"{self.start_emoji} Starting container {self.container_name}...")
            self.runtime.start_container(self.container_name)
            return True

        elif create_if_missing and "image" in run_options:
            # Container doesn't exist, create and start it
            self.console.print(
                f"{self.start_emoji} Creating and starting container {self.container_name}..."
            )

            image = run_options.pop("image")

            # Convert volume mappings dict to list format for Docker
            volumes_dict = run_options.get("volumes", {})
            volumes_list = None
            if isinstance(volumes_dict, dict):
                volumes_list = [f"{host}:{container}" for host, container in volumes_dict.items()]
            elif volumes_dict:
                volumes_list = volumes_dict  # Already a list

            self.runtime.run_container(
                image=image,
                name=self.container_name,
                detach=True,
                volumes=volumes_list,
                environment=run_options.get("environment"),
                ports=run_options.get("ports"),
                workdir=run_options.get("workdir"),
                command=run_options.get("command"),
            )

            self.console.print()
            return True

        else:
            self.console.print(f"[red]❌ Container {self.container_name} does not exist[/red]")
            return False

    except ProcessError as e:
        logger.error(
            "Failed to start container",
            name=self.container_name,
            error=str(e),
            stderr=e.stderr,
        )
        self.console.print(f"[red]❌ Failed to start container: {e}[/red]")
        return False

status

status() -> dict[str, Any]

Get container status.

Returns:

Type	Description
dict[str, Any]	Status information dictionary

Source code in wrknv/container/operations/lifecycle.py

def status(self) -> dict[str, Any]:
    """Get container status.

    Returns:
        Status information dictionary
    """
    try:
        exists = self.runtime.container_exists(self.container_name)
        running = False
        info = {}

        if exists:
            running = self.runtime.container_running(self.container_name)
            info = self.runtime.inspect_container(self.container_name)

        status = {
            "name": self.container_name,
            "exists": exists,
            "running": running,
            "status": "running" if running else ("stopped" if exists else "not found"),
        }

        # Add extra info if available
        if info:
            status["id"] = info.get("Id", "")[:12]
            status["image"] = info.get("Config", {}).get("Image", "")

            if info.get("State"):
                state = info["State"]
                status["started_at"] = state.get("StartedAt")
                status["finished_at"] = state.get("FinishedAt")

        return status

    except ProcessError as e:
        logger.error(
            "Failed to get container status",
            name=self.container_name,
            error=str(e),
        )
        return {
            "name": self.container_name,
            "exists": False,
            "running": False,
            "status": "error",
            "error": str(e),
        }

stop

stop(timeout: int) -> bool

Stop the container.

Parameters:

Name	Type	Description	Default
timeout	int	Seconds to wait before force stopping	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/lifecycle.py

def stop(self, timeout: int) -> bool:
    """Stop the container.

    Args:
        timeout: Seconds to wait before force stopping

    Returns:
        True if successful
    """
    try:
        if not self.runtime.container_running(self.container_name):
            self.console.print(f"[yellow]⚠️  Container {self.container_name} is not running[/yellow]")
            return True

        self.console.print(f"{self.stop_emoji} Stopping container {self.container_name}...")

        self.runtime.stop_container(self.container_name, timeout=timeout)

        return True

    except ProcessError as e:
        logger.error(
            "Failed to stop container",
            name=self.container_name,
            error=str(e),
        )
        self.console.print(f"[red]❌ Failed to stop container: {e}[/red]")
        return False

ContainerLogs

Handles container log operations.

Functions

clear_logs

clear_logs() -> bool

Clear container logs (if supported by runtime).

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/logs.py

def clear_logs(self) -> bool:
    """Clear container logs (if supported by runtime).

    Returns:
        True if successful
    """
    try:
        from provide.foundation.process import run

        # Docker doesn't have a direct clear logs command
        # This is a workaround using truncate
        run(
            [
                "sh",
                "-c",
                f"truncate -s 0 $(docker inspect --format='{{{{.LogPath}}}}' {self.container_name})",
            ],
            check=True,
        )

        logger.info("Container logs cleared", container=self.container_name)
        return True

    except ProcessError as e:
        logger.warning(
            "Failed to clear logs (may not be supported)",
            container=self.container_name,
            error=str(e),
        )
        self.console.print("[yellow]⚠️  Log clearing not supported or failed[/yellow]")
        return False

get_logs

get_logs(
    follow: bool,
    tail: int | None,
    since: str | None,
    timestamps: bool,
) -> str | None

Get container logs.

Parameters:

Name	Type	Description	Default
follow	bool	Follow log output	required
tail	int | None	Number of lines to tail	required
since	str | None	Show logs since timestamp (e.g., "2023-01-01T00:00:00")	required
timestamps	bool	Show timestamps	required

Returns:

Type	Description
str | None	Log output if not following, None if following

Source code in wrknv/container/operations/logs.py

def get_logs(
    self,
    follow: bool,
    tail: int | None,
    since: str | None,
    timestamps: bool,
) -> str | None:
    """Get container logs.

    Args:
        follow: Follow log output
        tail: Number of lines to tail
        since: Show logs since timestamp (e.g., "2023-01-01T00:00:00")
        timestamps: Show timestamps

    Returns:
        Log output if not following, None if following
    """
    try:
        if follow:
            # Stream logs
            for line in self.stream_logs(
                tail=tail,
                since=since,
                timestamps=timestamps,
            ):
                self.console.print(line, end="")
            return None
        else:
            # Get logs as string
            result = self.runtime.get_container_logs(
                name=self.container_name,
                follow=False,
                tail=tail,
                since=since,
            )
            return result.stdout

    except ProcessError as e:
        logger.error(
            "Failed to get logs",
            container=self.container_name,
            error=str(e),
        )
        self.console.print(f"[red]❌ Failed to get logs: {e}[/red]")
        return None

show_logs

show_logs(
    lines: int | None,
    since_minutes: int | None,
    grep: str | None,
) -> None

Show container logs with filtering.

Parameters:

Name	Type	Description	Default
lines	int | None	Number of recent lines to show	required
since_minutes	int | None	Show logs from last N minutes	required
grep	str | None	Filter logs by pattern	required

Source code in wrknv/container/operations/logs.py

def show_logs(
    self,
    lines: int | None,
    since_minutes: int | None,
    grep: str | None,
) -> None:
    """Show container logs with filtering.

    Args:
        lines: Number of recent lines to show
        since_minutes: Show logs from last N minutes
        grep: Filter logs by pattern
    """
    # Calculate since timestamp if needed
    since = None
    if since_minutes:
        since_time = provide_now() - timedelta(minutes=since_minutes)
        since = since_time.isoformat()

    # Get logs
    logs = self.get_logs(
        follow=False,
        tail=lines,
        since=since,
        timestamps=False,
    )

    if not logs:
        self.console.print(f"[yellow]No logs found for {self.container_name}[/yellow]")
        return

    # Filter if grep provided
    if grep:
        filtered_lines = []
        for line in logs.splitlines():
            if grep.lower() in line.lower():
                filtered_lines.append(line)
        logs = "\n".join(filtered_lines)

    # Display logs
    if logs:
        self.console.print(logs)
    else:
        self.console.print("[yellow]No matching logs found[/yellow]")

stream_logs

stream_logs(
    tail: int | None, since: str | None, timestamps: bool
) -> Generator[str, None, None]

Stream container logs.

Parameters:

Name	Type	Description	Default
tail	int | None	Number of lines to tail	required
since	str | None	Show logs since timestamp	required
timestamps	bool	Show timestamps	required

Yields:

Type	Description
str	Log lines

Source code in wrknv/container/operations/logs.py

def stream_logs(
    self,
    tail: int | None,
    since: str | None,
    timestamps: bool,
) -> Generator[str, None, None]:
    """Stream container logs.

    Args:
        tail: Number of lines to tail
        since: Show logs since timestamp
        timestamps: Show timestamps

    Yields:
        Log lines
    """
    cmd = [self.runtime.runtime_command, "logs", "-f"]

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

    cmd.append(self.container_name)

    try:
        yield from stream(cmd)
    except ProcessError as e:
        logger.error(
            "Failed to stream logs",
            container=self.container_name,
            error=str(e),
        )
        self.console.print(f"[red]❌ Log streaming failed: {e}[/red]")

VolumeManager

Manages container volume operations.

Functions

backup

backup(
    backup_path: Path | None = None,
    volumes: list[str] | None = None,
    compress: bool = True,
) -> bool

Convenience method for backing up volumes.

Parameters:

Name	Type	Description	Default
backup_path	Path | None	Path to save backup	None
volumes	list[str] | None	List of volume names to backup	None
compress	bool	Whether to compress the backup	True

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/volumes.py

def backup(
    self,
    backup_path: Path | None = None,
    volumes: list[str] | None = None,
    compress: bool = True,
) -> bool:
    """Convenience method for backing up volumes.

    Args:
        backup_path: Path to save backup
        volumes: List of volume names to backup
        compress: Whether to compress the backup

    Returns:
        True if successful
    """
    # For now, just return True as a stub
    # Full implementation would backup specified volumes
    return True

backup_volume

backup_volume(
    volume_name: str, container_name: str, mount_path: str
) -> Path | None

Backup a volume to a tar file.

Parameters:

Name	Type	Description	Default
volume_name	str	Volume to backup	required
container_name	str	Container using the volume	required
mount_path	str	Mount path inside container	required

Returns:

Type	Description
Path | None	Path to backup file if successful

Source code in wrknv/container/operations/volumes.py

def backup_volume(
    self,
    volume_name: str,
    container_name: str,
    mount_path: str,
) -> Path | None:
    """Backup a volume to a tar file.

    Args:
        volume_name: Volume to backup
        container_name: Container using the volume
        mount_path: Mount path inside container

    Returns:
        Path to backup file if successful
    """
    try:
        timestamp = provide_now().strftime("%Y%m%d_%H%M%S")
        backup_file = self.backup_dir / f"{volume_name}_{timestamp}.tar"

        self.console.print(f"[cyan]💾 Backing up volume {volume_name}...[/cyan]")

        # Create backup using tar inside container
        cmd = [
            self.runtime.runtime_command,
            "run",
            "--rm",
            "-v",
            f"{volume_name}:{mount_path}",
            "-v",
            f"{self.backup_dir}:/backup",
            "alpine",
            "tar",
            "-czf",
            f"/backup/{backup_file.name}",
            "-C",
            mount_path,
            ".",
        ]

        run(cmd, check=True)

        logger.info(
            "Volume backed up",
            volume=volume_name,
            backup_file=str(backup_file),
        )
        return backup_file

    except ProcessError as e:
        logger.error("Failed to backup volume", volume=volume_name, error=str(e))
        self.console.print(f"[red]❌ Backup failed: {e}[/red]")
        return None

clean

clean(preserve: list[str] | None = None) -> bool

Clean up volumes.

Parameters:

Name	Type	Description	Default
preserve	list[str] | None	List of volume names to preserve	None

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/volumes.py

def clean(self, preserve: list[str] | None = None) -> bool:
    """Clean up volumes.

    Args:
        preserve: List of volume names to preserve

    Returns:
        True if successful
    """
    # For now, just return True as a stub
    # Full implementation would clean up volumes
    return True

create_volume

create_volume(
    name: str,
    driver: str | None,
    options: dict[str, str] | None,
) -> bool

Create a named volume.

Parameters:

Name	Type	Description	Default
name	str	Volume name	required
driver	str | None	Volume driver (e.g., "local")	required
options	dict[str, str] | None	Driver options	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/volumes.py

def create_volume(self, name: str, driver: str | None, options: dict[str, str] | None) -> bool:
    """Create a named volume.

    Args:
        name: Volume name
        driver: Volume driver (e.g., "local")
        options: Driver options

    Returns:
        True if successful
    """
    try:
        cmd = [self.runtime.runtime_command, "volume", "create"]

        if driver:
            cmd.extend(["--driver", driver])

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

        cmd.append(name)

        run(cmd, check=True)

        logger.info("Volume created", name=name, driver=driver)
        return True

    except ProcessError as e:
        logger.error("Failed to create volume", name=name, error=str(e))
        self.console.print(f"[red]❌ Failed to create volume: {e}[/red]")
        return False

inspect_volume

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

Get detailed volume information.

Parameters:

Name	Type	Description	Default
name	str	Volume name	required

Returns:

Type	Description
dict[str, Any]	Volume information

Source code in wrknv/container/operations/volumes.py

def inspect_volume(self, name: str) -> dict[str, Any]:
    """Get detailed volume information.

    Args:
        name: Volume name

    Returns:
        Volume information
    """
    try:
        result = run([self.runtime.runtime_command, "volume", "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 volume", name=name, error=str(e))
        return {}

list_volumes

list_volumes(
    filter_label: str | None,
) -> list[dict[str, Any]]

List all volumes.

Parameters:

Name	Type	Description	Default
filter_label	str | None	Filter by label	required

Returns:

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

Source code in wrknv/container/operations/volumes.py

def list_volumes(self, filter_label: str | None) -> list[dict[str, Any]]:
    """List all volumes.

    Args:
        filter_label: Filter by label

    Returns:
        List of volume information
    """
    try:
        cmd = [self.runtime.runtime_command, "volume", "ls", "--format", "json"]

        if filter_label:
            cmd.extend(["--filter", f"label={filter_label}"])

        result = run(cmd, check=True)

        volumes = []
        if result.stdout:
            for line in result.stdout.strip().splitlines():
                if line:
                    volumes.append(json.loads(line))

        return volumes

    except (ProcessError, json.JSONDecodeError) as e:
        logger.error("Failed to list volumes", error=str(e))
        return []

remove_volume

remove_volume(name: str, force: bool) -> bool

Remove a named volume.

Parameters:

Name	Type	Description	Default
name	str	Volume name	required
force	bool	Force removal	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/volumes.py

def remove_volume(self, name: str, force: bool) -> bool:
    """Remove a named volume.

    Args:
        name: Volume name
        force: Force removal

    Returns:
        True if successful
    """
    try:
        cmd = [self.runtime.runtime_command, "volume", "rm"]

        if force:
            cmd.append("-f")

        cmd.append(name)

        run(cmd, check=True)

        logger.info("Volume removed", name=name)
        return True

    except ProcessError as e:
        logger.error("Failed to remove volume", name=name, error=str(e))
        self.console.print(f"[red]❌ Failed to remove volume: {e}[/red]")
        return False

restore

restore(backup_path: Path, force: bool = False) -> bool

Convenience method for restoring volumes.

Parameters:

Name	Type	Description	Default
backup_path	Path	Path to backup file	required
force	bool	Force restore even if volume exists	False

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/volumes.py

def restore(self, backup_path: Path, force: bool = False) -> bool:
    """Convenience method for restoring volumes.

    Args:
        backup_path: Path to backup file
        force: Force restore even if volume exists

    Returns:
        True if successful
    """
    # For now, just return True as a stub
    # Full implementation would restore from backup
    return True

restore_volume

restore_volume(
    volume_name: str, backup_file: Path, mount_path: str
) -> bool

Restore a volume from a tar file.

Parameters:

Name	Type	Description	Default
volume_name	str	Volume to restore to	required
backup_file	Path	Backup tar file	required
mount_path	str	Mount path inside container	required

Returns:

Type	Description
bool	True if successful

Source code in wrknv/container/operations/volumes.py

def restore_volume(
    self,
    volume_name: str,
    backup_file: Path,
    mount_path: str,
) -> bool:
    """Restore a volume from a tar file.

    Args:
        volume_name: Volume to restore to
        backup_file: Backup tar file
        mount_path: Mount path inside container

    Returns:
        True if successful
    """
    try:
        if not backup_file.exists():
            self.console.print(f"[red]❌ Backup file not found: {backup_file}[/red]")
            return False

        self.console.print(f"[cyan]📥 Restoring volume {volume_name}...[/cyan]")

        # Restore using tar inside container
        cmd = [
            self.runtime.runtime_command,
            "run",
            "--rm",
            "-v",
            f"{volume_name}:{mount_path}",
            "-v",
            f"{backup_file.parent}:/backup",
            "alpine",
            "tar",
            "-xzf",
            f"/backup/{backup_file.name}",
            "-C",
            mount_path,
        ]

        run(cmd, check=True)

        logger.info(
            "Volume restored",
            volume=volume_name,
            backup_file=str(backup_file),
        )
        return True

    except ProcessError as e:
        logger.error(
            "Failed to restore volume",
            volume=volume_name,
            backup_file=str(backup_file),
            error=str(e),
        )
        self.console.print(f"[red]❌ Restore failed: {e}[/red]")
        return False

show_volumes

show_volumes() -> None

Display volumes in a table.

Source code in wrknv/container/operations/volumes.py

def show_volumes(self) -> None:
    """Display volumes in a table."""
    volumes = self.list_volumes(filter_label=None)

    if not volumes:
        self.console.print("[yellow]No volumes found[/yellow]")
        return

    table = Table(title="Container Volumes")
    table.add_column("Name", style="cyan")
    table.add_column("Driver", style="green")
    table.add_column("Mountpoint", style="yellow")

    for volume in volumes:
        table.add_row(
            volume.get("Name", ""),
            volume.get("Driver", ""),
            volume.get("Mountpoint", "")[:50] + "..."
            if len(volume.get("Mountpoint", "")) > 50
            else volume.get("Mountpoint", ""),
        )

    self.console.print(table)