tessl/pypi-docker-py

A comprehensive Python client library for the Docker Remote API enabling programmatic control of containers, images, networks, and volumes.

—

Pending

Overview

Eval results

Files

Services and Swarm

Name: tessl/pypi-docker-py
Author: tessl

Docker Swarm cluster management and service orchestration capabilities. Provides container orchestration, service scaling, rolling updates, cluster administration, and high availability across multiple Docker hosts with built-in load balancing and service discovery.

Capabilities

Swarm Cluster Management

Functions for initializing, joining, and managing Docker Swarm clusters.

def init_swarm(advertise_addr=None, listen_addr='0.0.0.0:2377', 
               force_new_cluster=False, swarm_spec=None):
    """
    Initialize a new Swarm cluster.
    
    Parameters:
    - advertise_addr (str): IP address to advertise to other nodes
    - listen_addr (str): Address to listen for cluster management traffic
    - force_new_cluster (bool): Force creation of new cluster
    - swarm_spec (SwarmSpec): Swarm configuration specification
    
    Returns:
    dict: Swarm initialization response with node ID and tokens
    """

def join_swarm(remote_addrs, join_token, listen_addr=None, advertise_addr=None):
    """
    Join an existing Swarm cluster.
    
    Parameters:
    - remote_addrs (list): List of manager node addresses
    - join_token (str): Token for joining (worker or manager token)
    - listen_addr (str): Address to listen for cluster traffic
    - advertise_addr (str): IP address to advertise to cluster
    
    Returns:
    None
    """

def leave_swarm(force=False):
    """
    Leave the Swarm cluster.
    
    Parameters:
    - force (bool): Force leave even if node is a manager
    
    Returns:
    None
    """

def inspect_swarm():
    """
    Get detailed information about the Swarm cluster.
    
    Returns:
    dict: Swarm configuration and state information
    """

def update_swarm(version, swarm_spec=None, rotate_worker_token=False, 
                 rotate_manager_token=False):
    """
    Update Swarm cluster configuration.
    
    Parameters:
    - version (int): Current version of swarm object being updated
    - swarm_spec (SwarmSpec): New swarm configuration
    - rotate_worker_token (bool): Rotate worker join token
    - rotate_manager_token (bool): Rotate manager join token
    
    Returns:
    None
    """

Node Management

Functions for managing Swarm cluster nodes.

def nodes(filters=None):
    """
    List Swarm nodes.
    
    Parameters:
    - filters (dict): Filters to apply to node list
    
    Returns:
    list: List of node dictionaries with details and status
    """

def inspect_node(node_id):
    """
    Get detailed information about a Swarm node.
    
    Parameters:
    - node_id (str): Node ID or name
    
    Returns:
    dict: Node configuration, status, and resource information
    """

Service Management

Functions for creating, managing, and scaling Docker services in Swarm mode.

def services(filters=None):
    """
    List Swarm services.
    
    Parameters:
    - filters (dict): Filters to apply to service list
    
    Returns:
    list: List of service dictionaries with configuration and status
    """

def create_service(task_template, name=None, labels=None, mode=None, 
                  update_config=None, networks=None, endpoint_config=None):
    """
    Create a new Swarm service.
    
    Parameters:
    - task_template (TaskTemplate): Service task specification
    - name (str): Service name
    - labels (dict): Service labels
    - mode (dict): Service mode configuration (replicated/global)
    - update_config (UpdateConfig): Update configuration
    - networks (list): Networks to attach service to
    - endpoint_config (dict): Endpoint configuration for service ports
    
    Returns:
    dict: Service creation response with service ID
    """

def inspect_service(service):
    """
    Get detailed information about a service.
    
    Parameters:
    - service (str): Service ID or name
    
    Returns:
    dict: Service specification and current state
    """

def update_service(service, version, task_template=None, name=None, 
                  labels=None, mode=None, update_config=None, 
                  networks=None, endpoint_config=None):
    """
    Update an existing service.
    
    Parameters:
    - service (str): Service ID or name
    - version (int): Current version of service object being updated
    - task_template (TaskTemplate): New task specification
    - name (str): New service name
    - labels (dict): New service labels
    - mode (dict): New service mode configuration
    - update_config (UpdateConfig): New update configuration
    - networks (list): New network attachments
    - endpoint_config (dict): New endpoint configuration
    
    Returns:
    dict: Service update response
    """

def remove_service(service):
    """
    Remove a service.
    
    Parameters:
    - service (str): Service ID or name
    
    Returns:
    None
    """

Task Management

Functions for inspecting and managing service tasks.

def tasks(filters=None):
    """
    List Swarm tasks.
    
    Parameters:
    - filters (dict): Filters to apply to task list
    
    Returns:
    list: List of task dictionaries with status and assignment
    """

def inspect_task(task):
    """
    Get detailed information about a task.
    
    Parameters:
    - task (str): Task ID
    
    Returns:
    dict: Task specification, status, and runtime information
    """

Volume Management

Docker volume operations for persistent data storage in Swarm services.

def volumes(filters=None):
    """
    List Docker volumes.
    
    Parameters:
    - filters (dict): Filters to apply to volume list
    
    Returns:
    list: List of volume dictionaries
    """

def create_volume(name, driver=None, driver_opts=None, labels=None):
    """
    Create a Docker volume.
    
    Parameters:
    - name (str): Volume name
    - driver (str): Volume driver
    - driver_opts (dict): Driver-specific options
    - labels (dict): Volume labels
    
    Returns:
    dict: Volume creation response
    """

def inspect_volume(name):
    """
    Get detailed information about a volume.
    
    Parameters:
    - name (str): Volume name
    
    Returns:
    dict: Volume configuration and status
    """

def remove_volume(name):
    """
    Remove a volume.
    
    Parameters:
    - name (str): Volume name
    
    Returns:
    None
    """

Service Configuration Types

TaskTemplate

Service task specification defining container configuration and resource requirements.

class TaskTemplate:
    def __init__(self, container_spec, resources=None, restart_policy=None, 
                 placement=None, log_driver=None):
        """
        Create task template for service.
        
        Parameters:
        - container_spec (ContainerSpec): Container specification
        - resources (Resources): Resource constraints
        - restart_policy (RestartPolicy): Restart policy configuration
        - placement (dict): Task placement constraints
        - log_driver (dict): Logging driver configuration
        """

    # Properties
    container_spec: ContainerSpec  # Container configuration
    resources: Resources           # Resource limits and reservations
    restart_policy: RestartPolicy  # Restart behavior
    placement: dict               # Node placement preferences

ContainerSpec

Container specification for services defining image, command, and runtime configuration.

class ContainerSpec:
    def __init__(self, image, command=None, args=None, env=None, 
                 workdir=None, user=None, labels=None, mounts=None, 
                 stop_grace_period=None):
        """
        Create container specification for service tasks.
        
        Parameters:
        - image (str): Container image
        - command (list): Command to run
        - args (list): Command arguments
        - env (list): Environment variables
        - workdir (str): Working directory
        - user (str): User to run as
        - labels (dict): Container labels
        - mounts (list): Volume mounts
        - stop_grace_period (int): Grace period before force kill
        """

Resources

Resource constraints for service tasks including CPU and memory limits.

class Resources:
    def __init__(self, cpu_limit=None, mem_limit=None, cpu_reservation=None, 
                 mem_reservation=None):
        """
        Create resource constraints for service tasks.
        
        Parameters:
        - cpu_limit (int): CPU limit in nanocpus
        - mem_limit (int): Memory limit in bytes
        - cpu_reservation (int): CPU reservation in nanocpus
        - mem_reservation (int): Memory reservation in bytes
        """

UpdateConfig

Configuration for service rolling updates and deployment strategy.

class UpdateConfig:
    def __init__(self, parallelism=0, delay=None, failure_action='continue'):
        """
        Create update configuration for service updates.
        
        Parameters:
        - parallelism (int): Number of tasks to update simultaneously
        - delay (int): Delay between task updates in nanoseconds
        - failure_action (str): Action on update failure ('continue', 'pause')
        """

RestartPolicy

Container restart policy configuration for service tasks.

class RestartPolicy:
    def __init__(self, condition='none', delay=0, max_attempts=0, window=0):
        """
        Create restart policy for service tasks.
        
        Parameters:
        - condition (str): When to restart ('none', 'on-failure', 'any')
        - delay (int): Delay between restart attempts in nanoseconds
        - max_attempts (int): Maximum restart attempts
        - window (int): Window to evaluate restart policy
        """

# Restart condition constants
class RestartConditionTypesEnum:
    NONE = 'none'
    ON_FAILURE = 'on-failure'  
    ANY = 'any'

SwarmSpec

Swarm cluster configuration specification.

class SwarmSpec:
    def __init__(self, task_history_retention_limit=None, snapshot_interval=None, 
                 keep_old_snapshots=None, log_entries_for_slow_followers=None, 
                 heartbeat_tick=None, election_tick=None, 
                 dispatcher_heartbeat_period=None, node_cert_expiry=None, 
                 external_ca=None, name=None):
        """
        Create Swarm cluster specification.
        
        Parameters:
        - task_history_retention_limit (int): Task history retention limit
        - snapshot_interval (int): Raft snapshot interval
        - keep_old_snapshots (int): Old snapshots to keep
        - log_entries_for_slow_followers (int): Log entries for slow followers
        - heartbeat_tick (int): Heartbeat tick interval
        - election_tick (int): Election tick interval
        - dispatcher_heartbeat_period (int): Dispatcher heartbeat period
        - node_cert_expiry (int): Node certificate expiry time
        - external_ca (SwarmExternalCA): External certificate authority
        - name (str): Swarm cluster name
        """

def create_swarm_spec(*args, **kwargs):
    """
    Create SwarmSpec object with given parameters.
    
    Returns:
    SwarmSpec: Swarm specification object
    """

SwarmExternalCA

External certificate authority configuration for Swarm cluster security.

class SwarmExternalCA:
    def __init__(self, url, protocol=None, options=None):
        """
        Create external CA configuration.
        
        Parameters:
        - url (str): CA server URL
        - protocol (str): Protocol to use
        - options (dict): Additional CA options
        """

Usage Examples

Basic Swarm Setup

import docker
from docker.types import TaskTemplate, ContainerSpec, Resources

client = docker.Client()

# Initialize Swarm cluster
swarm_info = client.init_swarm(
    advertise_addr='192.168.1.100',
    listen_addr='0.0.0.0:2377'
)

print(f"Swarm initialized with Node ID: {swarm_info['NodeID']}")
print(f"Worker token: {swarm_info['Tokens']['Worker']}")
print(f"Manager token: {swarm_info['Tokens']['Manager']}")

# Get Swarm information
swarm_details = client.inspect_swarm()
print(f"Swarm ID: {swarm_details['ID']}")
print(f"Node count: {len(client.nodes())}")

Creating and Managing Services

# Create container specification
container_spec = ContainerSpec(
    image='nginx:latest',
    env=['NGINX_PORT=80'],
    mounts=[
        {
            'Type': 'volume',
            'Source': 'nginx-data',
            'Target': '/usr/share/nginx/html'
        }
    ]
)

# Create resource constraints  
resources = Resources(
    cpu_limit=1000000000,  # 1 CPU (1 billion nanocpus)
    mem_limit=512 * 1024 * 1024,  # 512MB
    cpu_reservation=500000000,  # 0.5 CPU reservation
    mem_reservation=256 * 1024 * 1024  # 256MB reservation
)

# Create task template
task_template = TaskTemplate(
    container_spec=container_spec,
    resources=resources,
    restart_policy=docker.types.RestartPolicy(
        condition='on-failure',
        max_attempts=3
    )
)

# Create service
service = client.create_service(
    task_template=task_template,
    name='web-service',
    mode={'Replicated': {'Replicas': 3}},  # 3 replicas
    endpoint_config={
        'Ports': [
            {
                'Protocol': 'tcp',
                'PublishedPort': 80,
                'TargetPort': 80
            }
        ]
    },
    labels={'app': 'web', 'version': '1.0'}
)

print(f"Created service: {service['ID']}")

Service Scaling and Updates

# List services
services = client.services()
for service in services:
    print(f"Service: {service['Spec']['Name']}, Replicas: {service['Spec']['Mode']['Replicated']['Replicas']}")

# Update service to scale to 5 replicas
service_info = client.inspect_service('web-service')
version = service_info['Version']['Index']

client.update_service(
    'web-service',
    version=version,
    mode={'Replicated': {'Replicas': 5}}
)

# Update service with new image
updated_container_spec = ContainerSpec(
    image='nginx:1.21-alpine',  # New image version
    env=['NGINX_PORT=80']
)

updated_task_template = TaskTemplate(
    container_spec=updated_container_spec,
    resources=resources
)

service_info = client.inspect_service('web-service')
version = service_info['Version']['Index']

client.update_service(
    'web-service',
    version=version,
    task_template=updated_task_template,
    update_config=docker.types.UpdateConfig(
        parallelism=1,  # Update one task at a time
        delay=5000000000,  # 5 second delay between updates
        failure_action='pause'
    )
)

Advanced Service Configuration

# Create service with complex configuration
container_spec = ContainerSpec(
    image='myapp:latest',
    command=['python', 'app.py'],
    env=[
        'DATABASE_URL=postgresql://user:pass@db:5432/myapp',
        'REDIS_URL=redis://cache:6379'
    ],
    labels={'service': 'api'},
    mounts=[
        {
            'Type': 'volume',
            'Source': 'app-data',
            'Target': '/app/data'
        }
    ],
    user='1000:1000',
    workdir='/app'
)

# Create with placement constraints
task_template = TaskTemplate(
    container_spec=container_spec,
    resources=Resources(
        cpu_limit=2000000000,  # 2 CPUs
        mem_limit=1024 * 1024 * 1024,  # 1GB
        cpu_reservation=1000000000,  # 1 CPU reserved
        mem_reservation=512 * 1024 * 1024  # 512MB reserved
    ),
    placement={
        'Constraints': ['node.role==worker'],
        'Preferences': [
            {
                'Spread': {'SpreadDescriptor': 'node.labels.zone'}
            }
        ]
    },
    restart_policy=docker.types.RestartPolicy(
        condition='any',
        delay=5000000000,  # 5 seconds
        max_attempts=5
    )
)

# Create service with networks
service = client.create_service(
    task_template=task_template,
    name='api-service',
    mode={'Replicated': {'Replicas': 3}},
    networks=[
        {'Target': 'frontend-network'},
        {'Target': 'backend-network'}
    ],
    endpoint_config={
        'Ports': [
            {
                'Protocol': 'tcp',
                'PublishedPort': 8000,
                'TargetPort': 8000,
                'PublishMode': 'ingress'
            }
        ]
    },
    labels={
        'app': 'myapp',
        'tier': 'api',
        'environment': 'production'
    }
)

Service Monitoring and Management

# Monitor service tasks
tasks = client.tasks(filters={'service': 'web-service'})
for task in tasks:
    print(f"Task: {task['ID'][:12]}")
    print(f"  State: {task['Status']['State']}")
    print(f"  Node: {task['NodeID'][:12]}")
    if 'ContainerStatus' in task['Status']:
        print(f"  Container: {task['Status']['ContainerStatus']['ContainerID'][:12]}")

# Get detailed task information
if tasks:
    task_detail = client.inspect_task(tasks[0]['ID'])
    print(f"Task created: {task_detail['CreatedAt']}")
    print(f"Task updated: {task_detail['UpdatedAt']}")

# Service logs (requires newer API versions)
try:
    logs = client.logs('web-service', stdout=True, stderr=True)
    print("Service logs:", logs.decode('utf-8')[:200])
except Exception as e:
    print(f"Logs not available: {e}")

# Remove service
client.remove_service('web-service')

Node Management

# List all nodes in the swarm
nodes = client.nodes()
for node in nodes:
    print(f"Node: {node['Description']['Hostname']}")
    print(f"  ID: {node['ID'][:12]}")
    print(f"  Role: {node['Spec']['Role']}")
    print(f"  Status: {node['Status']['State']}")
    print(f"  Availability: {node['Spec']['Availability']}")

# Get detailed node information
if nodes:
    node_detail = client.inspect_node(nodes[0]['ID'])
    resources = node_detail['Description']['Resources']
    print(f"Node Resources:")
    print(f"  CPUs: {resources['NanoCPUs'] / 1000000000}")
    print(f"  Memory: {resources['MemoryBytes'] / (1024**3):.2f} GB")

Swarm Leave and Cleanup

# Leave swarm (on worker node)
try:
    client.leave_swarm()
    print("Left swarm successfully")
except Exception as e:
    print(f"Error leaving swarm: {e}")

# Force leave swarm (on manager node)
try:
    client.leave_swarm(force=True)
    print("Force left swarm successfully")  
except Exception as e:
    print(f"Error force leaving swarm: {e}")

Install with Tessl CLI