System Operations

def info() -> SystemInfo:
    """
    Get system-wide information about Docker.
    
    Returns:
    - SystemInfo object with detailed system information
    """

def df() -> DiskFreeResult:
    """
    Show docker filesystem usage.
    
    Returns:
    - DiskFreeResult with breakdown of space usage by type
    """

def prune(
    *,
    all: bool = False,
    volumes: bool = False,
    filters: Optional[Dict[str, str]] = None,
    force: bool = False
) -> Dict[str, Any]:
    """
    Remove unused Docker objects system-wide.
    
    Parameters:
    - all: Remove all unused objects, not just dangling ones
    - volumes: Also remove unused volumes
    - filters: Filters to apply to cleanup
    - force: Don't prompt for confirmation
    
    Returns:
    - Information about removed objects and space reclaimed
    """

def events(
    *,
    since: Optional[str] = None,
    until: Optional[str] = None,
    filters: Optional[Dict[str, str]] = None,
    format: Optional[str] = None
) -> Iterator[Dict[str, Any]]:
    """
    Get real-time events from Docker daemon.
    
    Parameters:
    - since: Show events since timestamp
    - until: Show events until timestamp
    - filters: Filter events (container, image, volume, network, daemon, etc.)
    - format: Output format template
    
    Returns:
    - Iterator of event dictionaries
    """

from python_on_whales import docker

# Get comprehensive system information
info = docker.system.info()
print(f"Docker Version: {info.server_version}")
print(f"Architecture: {info.architecture}")
print(f"Operating System: {info.operating_system}")
print(f"Total Memory: {info.mem_total}")
print(f"Containers: {info.containers} (Running: {info.containers_running})")
print(f"Images: {info.images}")

# Check disk usage
usage = docker.system.df()
print(f"Images: {len(usage.images)} using {usage.images_size}")
print(f"Containers: {len(usage.containers)} using {usage.containers_size}")
print(f"Volumes: {len(usage.volumes)} using {usage.volumes_size}")
print(f"Build Cache: {usage.build_cache_size}")

# Perform comprehensive cleanup
cleanup_result = docker.system.prune(
    all=True,
    volumes=True,
    force=True
)

print(f"Containers removed: {cleanup_result.get('ContainersDeleted', 0)}")
print(f"Networks removed: {cleanup_result.get('NetworksDeleted', 0)}")
print(f"Images removed: {cleanup_result.get('ImagesDeleted', 0)}")
print(f"Volumes removed: {cleanup_result.get('VolumesDeleted', 0)}")
print(f"Space reclaimed: {cleanup_result.get('SpaceReclaimed', 0)}")

# Targeted cleanup with filters
filtered_cleanup = docker.system.prune(
    filters={"until": "24h"},
    force=True
)

import json
from datetime import datetime

# Monitor all events
print("Monitoring Docker events (Ctrl+C to stop)...")
try:
    for event in docker.system.events():
        timestamp = datetime.fromtimestamp(event['time'])
        print(f"[{timestamp}] {event['Type']}: {event['Action']} - {event.get('Actor', {}).get('Attributes', {}).get('name', 'N/A')}")
except KeyboardInterrupt:
    print("Stopped monitoring events")

# Monitor specific event types
container_events = docker.system.events(
    filters={"type": "container", "event": ["start", "stop", "die"]}
)

for event in container_events:
    container_name = event.get('Actor', {}).get('Attributes', {}).get('name', 'unknown')
    print(f"Container {container_name}: {event['Action']}")
    
    # Break after 10 events for demo
    if event['time'] > (datetime.now().timestamp() + 60):
        break

def check_docker_health():
    """Comprehensive Docker system health check."""
    try:
        info = docker.system.info()
        
        # Check if daemon is responsive
        print("✓ Docker daemon is responsive")
        
        # Check system resources
        if info.mem_total < 2 * 1024 * 1024 * 1024:  # Less than 2GB
            print("⚠ Warning: Low system memory")
        else:
            print("✓ System memory looks good")
            
        # Check storage driver
        print(f"✓ Storage driver: {info.storage_driver}")
        
        # Check for warnings
        if hasattr(info, 'warnings') and info.warnings:
            print("⚠ System warnings:")
            for warning in info.warnings:
                print(f"  - {warning}")
        else:
            print("✓ No system warnings")
            
        # Check disk usage
        usage = docker.system.df()
        total_size = sum([
            usage.images_size or 0,
            usage.containers_size or 0,
            usage.volumes_size or 0,
            usage.build_cache_size or 0
        ])
        
        print(f"✓ Total Docker disk usage: {total_size / (1024**3):.2f} GB")
        
        return True
        
    except Exception as e:
        print(f"✗ Docker health check failed: {e}")
        return False

# Run health check
if check_docker_health():
    print("Docker system is healthy!")
else:
    print("Docker system has issues that need attention.")

class SystemInfo:
    id: str
    containers: int
    containers_running: int
    containers_paused: int
    containers_stopped: int
    images: int
    storage_driver: str
    logging_driver: str
    cgroup_driver: str
    cgroup_version: str
    plugins: Dict[str, List[str]]
    memory_limit: bool
    swap_limit: bool
    kernel_memory: bool
    oom_kill_disable: bool
    cpu_cfs_period: bool
    cpu_cfs_quota: bool
    cpu_shares: bool
    cpuset: bool
    pids_limit: bool
    ipv4_forwarding: bool
    bridge_nf_iptables: bool
    bridge_nf_ip6tables: bool
    debug: bool
    n_fd: int
    oom_score_adj: int
    n_goroutines: int
    system_time: str
    logging_driver: str
    cgroup_driver: str
    n_events_listener: int
    kernel_version: str
    operating_system: str
    os_type: str
    architecture: str
    index_server_address: str
    registry_config: Dict[str, Any]
    ncpu: int
    mem_total: int
    generic_resources: List[Dict[str, Any]]
    docker_root_dir: str
    http_proxy: str
    https_proxy: str
    no_proxy: str
    name: str
    labels: List[str]
    experimental_build: bool
    server_version: str
    cluster_store: str
    cluster_advertise: str
    runtimes: Dict[str, Dict[str, Any]]
    default_runtime: str
    swarm: Dict[str, Any]
    live_restore_enabled: bool
    isolation: str
    init_binary: str
    containerd_commit: Dict[str, str]
    runc_commit: Dict[str, str]
    init_commit: Dict[str, str]
    security_options: List[str]
    product_license: str
    warnings: Optional[List[str]]

class DiskFreeResult:
    layers_size: Optional[int]
    images: List[Dict[str, Any]]
    images_size: Optional[int]
    containers: List[Dict[str, Any]]
    containers_size: Optional[int]
    volumes: List[Dict[str, Any]]
    volumes_size: Optional[int]
    build_cache: List[Dict[str, Any]]
    build_cache_size: Optional[int]

class DockerEvent:
    type: str
    action: str
    time: int
    time_nano: int
    actor: Dict[str, Any]
    scope: Optional[str]