Task Management

def list() -> List[Task]:
    """
    List all tasks in the swarm.
    
    Returns:
    List of Task objects with status and assignment information
    """

def inspect(x: Union[str, List[str]]) -> Union[Task, List[Task]]:
    """
    Inspect one or more tasks.
    
    Parameters:
    - x: Task ID(s) - tasks are identified by their unique IDs
    
    Returns:
    Task object(s) with detailed information
    """

from python_on_whales import docker

# List all tasks in the swarm
tasks = docker.task.list()
for task in tasks:
    print(f"Task: {task.name}")
    print(f"  Service: {task.service_id}")
    print(f"  Node: {task.node_id}")
    print(f"  State: {task.desired_state}")
    print(f"  Status: {task.status.state}")
    print(f"  Slot: {task.slot}")

# Inspect specific task
task_id = "ktnakcyq2qg0kpgvjkb"
task = docker.task.inspect(task_id)
print(f"Task Details:")
print(f"  Created: {task.created_at}")
print(f"  Updated: {task.updated_at}")
print(f"  Image: {task.spec.container_spec.image}")
print(f"  Resources: {task.spec.resources}")

# Filter tasks by service (using service.ps() method)
web_service_tasks = docker.service.ps("web-service")
for task in web_service_tasks:
    print(f"Web Service Task: {task.name} - Status: {task.status.state}")

# Filter tasks by node (using node.ps() method)  
node_tasks = docker.node.ps("worker-node-01")
for task in node_tasks:
    print(f"Node Task: {task.name} - Service: {task.service_id}")

class Task:
    id: str
    version: DockerObjectVersion
    created_at: datetime
    updated_at: datetime
    name: str
    labels: Dict[str, str]
    spec: TaskSpec
    service_id: str
    slot: Optional[int]
    node_id: str
    assigned_generic_resources: List[Dict[str, Any]]
    status: TaskStatus
    desired_state: str

class TaskSpec:
    container_spec: ContainerSpec
    resources: Optional[ResourceRequirements]
    restart_policy: Optional[RestartPolicy]
    placement: Optional[Placement]
    log_driver: Optional[LogDriver]
    networks: List[NetworkAttachmentConfig]
    force_update: int

class TaskStatus:
    timestamp: datetime
    state: str  # "new", "assigned", "accepted", "preparing", "ready", "starting", "running", "complete", "shutdown", "failed", "rejected", "remove", "orphaned"
    message: Optional[str]
    err: Optional[str]
    container_status: Optional[ContainerStatus]
    port_status: Optional[PortStatus]

class ContainerStatus:
    container_id: str
    pid: int
    exit_code: Optional[int]

class ContainerSpec:
    image: str
    labels: Dict[str, str]
    command: Optional[List[str]]
    args: Optional[List[str]]
    hostname: Optional[str]
    env: Optional[List[str]]
    dir: Optional[str]
    user: Optional[str]
    groups: Optional[List[str]]
    privileges: Optional[Dict[str, Any]]
    tty: bool
    open_stdin: bool
    read_only: bool
    mounts: List[Mount]
    stop_signal: Optional[str]
    stop_grace_period: Optional[int]
    health_check: Optional[HealthConfig]
    hosts: Optional[List[str]]
    dns_config: Optional[DNSConfig]
    secrets: Optional[List[SecretReference]]
    configs: Optional[List[ConfigReference]]
    ulimits: Optional[List[Ulimit]]
    sysctls: Optional[Dict[str, str]]

class ResourceRequirements:
    limits: Optional[Resources]
    reservations: Optional[Resources]

class Resources:
    nano_cpus: Optional[int]
    memory_bytes: Optional[int]
    generic_resources: Optional[List[GenericResource]]

class RestartPolicy:
    condition: str  # "none", "on-failure", "any"
    delay: Optional[int]
    max_attempts: Optional[int]
    window: Optional[int]

class Placement:
    constraints: Optional[List[str]]
    preferences: Optional[List[PlacementPreference]]
    max_replicas: Optional[int]
    platforms: Optional[List[Platform]]

class LogDriver:
    name: str
    options: Optional[Dict[str, str]]

class DockerObjectVersion:
    index: int