A comprehensive Python client library for the Docker Remote API enabling programmatic control of containers, images, networks, and volumes.
—
Comprehensive utility functions, type classes, and configuration helpers that support Docker-py's core functionality. Includes data conversion utilities, authentication management, TLS configuration, and transport adapters for specialized connection types.
Helper functions for building complex Docker API configurations with proper validation and formatting.
def create_host_config(binds=None, port_bindings=None, lxc_conf=None,
publish_all_ports=False, links=None, privileged=False,
dns=None, dns_search=None, volumes_from=None,
network_mode=None, restart_policy=None, cap_add=None,
cap_drop=None, devices=None, extra_hosts=None,
read_only=None, pid_mode=None, ipc_mode=None,
security_opt=None, ulimits=None, log_config=None,
mem_limit=None, memswap_limit=None, cpu_shares=None,
cpuset_cpus=None, cpuset_mems=None, **kwargs):
"""
Create host configuration for containers.
Parameters:
- binds (dict|list): Volume bindings
- port_bindings (dict): Port bindings mapping
- lxc_conf (dict): LXC configuration options
- publish_all_ports (bool): Publish all exposed ports to random host ports
- links (list): Container links (deprecated)
- privileged (bool): Run container in privileged mode
- dns (list): Custom DNS servers
- dns_search (list): Custom DNS search domains
- volumes_from (list): Mount volumes from other containers
- network_mode (str): Network mode ('bridge', 'host', 'none', etc.)
- restart_policy (dict): Restart policy configuration
- cap_add (list): Linux capabilities to add
- cap_drop (list): Linux capabilities to drop
- devices (list): Device mappings
- extra_hosts (list): Extra host entries for /etc/hosts
- read_only (bool): Mount root filesystem as read-only
- pid_mode (str): PID namespace mode
- ipc_mode (str): IPC namespace mode
- security_opt (list): Security options
- ulimits (list): Resource limits (Ulimit objects)
- log_config (LogConfig): Logging configuration
- mem_limit (int|str): Memory limit
- memswap_limit (int|str): Memory + swap limit
- cpu_shares (int): CPU shares (relative weight)
- cpuset_cpus (str): CPUs to use
- cpuset_mems (str): Memory nodes to use
Returns:
dict: Host configuration dictionary
"""
def create_container_config(image, command=None, hostname=None, user=None,
detach=False, stdin_open=False, tty=False,
mem_limit=None, ports=None, environment=None,
dns=None, volumes=None, volumes_from=None,
network_disabled=False, name=None, entrypoint=None,
cpu_shares=None, working_dir=None, domainname=None,
memswap_limit=None, cpuset=None, host_config=None,
mac_address=None, labels=None, volume_driver=None,
stop_signal=None, networking_config=None, **kwargs):
"""
Create container configuration dictionary.
Parameters:
- image (str): Image name or ID
- command (str|list): Command to run
- hostname (str): Container hostname
- user (str): Username or UID
- detach (bool): Run in detached mode
- stdin_open (bool): Keep STDIN open
- tty (bool): Allocate pseudo-TTY
- mem_limit (int|str): Memory limit
- ports (list): Ports to expose
- environment (dict|list): Environment variables
- dns (list): DNS servers
- volumes (list): Volume mount points
- volumes_from (list): Volumes from other containers
- network_disabled (bool): Disable networking
- name (str): Container name
- entrypoint (str|list): Override default entrypoint
- cpu_shares (int): CPU shares
- working_dir (str): Working directory
- domainname (str): Domain name
- memswap_limit (int): Memory + swap limit
- cpuset (str): CPUs to use
- host_config (dict): Host configuration
- mac_address (str): MAC address
- labels (dict): Container labels
- volume_driver (str): Volume driver
- stop_signal (str): Stop signal
- networking_config (dict): Network configuration
Returns:
dict: Container configuration dictionary
"""
def create_networking_config(endpoints_config=None):
"""
Create networking configuration for containers.
Parameters:
- endpoints_config (dict): Endpoint configurations by network name
Returns:
dict: Networking configuration dictionary
"""
def create_endpoint_config(version, aliases=None, links=None,
ipv4_address=None, ipv6_address=None,
link_local_ips=None):
"""
Create endpoint configuration for network connections.
Parameters:
- version (str): API version
- aliases (list): Network-scoped aliases
- links (list): Container links (deprecated)
- ipv4_address (str): Static IPv4 address
- ipv6_address (str): Static IPv6 address
- link_local_ips (list): Link-local IP addresses
Returns:
dict: Endpoint configuration dictionary
"""Functions for converting data between different formats required by the Docker API.
def convert_port_bindings(port_bindings):
"""
Convert port bindings to Docker API format.
Parameters:
- port_bindings (dict): Port bindings mapping
Returns:
dict: Formatted port bindings for API
"""
def convert_volume_binds(binds):
"""
Convert volume bindings to Docker API format.
Parameters:
- binds (dict|list): Volume bindings
Returns:
list: Formatted volume bindings for API
"""
def convert_filters(filters):
"""
Convert filter dictionary to Docker API format.
Parameters:
- filters (dict): Filter conditions
Returns:
str: JSON-encoded filters for API
"""
def format_environment(environment):
"""
Format environment variables for Docker API.
Parameters:
- environment (dict|list): Environment variables
Returns:
list: Formatted environment variables
"""Functions for parsing various Docker-related data formats and specifications.
def parse_repository_tag(repo_name):
"""
Parse repository name and tag from image specification.
Parameters:
- repo_name (str): Repository name with optional tag
Returns:
tuple: (repository, tag) where tag defaults to 'latest'
"""
def parse_host(addr, is_win32=False, tls=False):
"""
Parse Docker host address into connection components.
Parameters:
- addr (str): Docker host address (tcp://, unix://, npipe://)
- is_win32 (bool): Windows platform flag
- tls (bool): TLS enabled flag
Returns:
tuple: (protocol, host, port, base_url)
"""
def parse_devices(devices):
"""
Parse device specifications into Docker API format.
Parameters:
- devices (list): Device specifications
Returns:
list: Formatted device mappings
"""
def parse_bytes(s):
"""
Parse byte size string into integer bytes.
Parameters:
- s (str): Size string (e.g., '512m', '1g', '2048k')
Returns:
int: Size in bytes
"""
def parse_env_file(env_file):
"""
Parse environment file into environment variables.
Parameters:
- env_file (str): Path to environment file
Returns:
dict: Environment variables from file
"""
def split_command(command):
"""
Split command string into array of arguments.
Parameters:
- command (str): Command string
Returns:
list: Command arguments array
"""
def normalize_links(links):
"""
Normalize container links to standard format.
Parameters:
- links (list): Container links
Returns:
list: Normalized link specifications
"""Functions for creating build contexts, tar archives, and handling Dockerfiles.
def mkbuildcontext(dockerfile):
"""
Create build context from Dockerfile content.
Parameters:
- dockerfile (str): Dockerfile content
Returns:
file: File-like object containing build context
"""
def tar(path, exclude=None, dockerfile=None, fileobj=None, gzip=False):
"""
Create tar archive from directory path.
Parameters:
- path (str): Directory path to archive
- exclude (list): Patterns to exclude (from .dockerignore)
- dockerfile (str): Custom Dockerfile content
- fileobj (file): File-like object to write to
- gzip (bool): Use gzip compression
Returns:
file: Tar archive file-like object
"""
def exclude_paths(root, patterns, dockerfile=None):
"""
Get list of paths that should be excluded from build context.
Parameters:
- root (str): Root directory path
- patterns (list): Exclusion patterns from .dockerignore
- dockerfile (str): Dockerfile name
Returns:
set: Set of paths to exclude
"""Functions for comparing Docker API versions.
def compare_version(v1, v2):
"""
Compare two Docker API version strings.
Parameters:
- v1 (str): First version
- v2 (str): Second version
Returns:
int: -1 if v1 < v2, 0 if equal, 1 if v1 > v2
"""
def version_lt(v1, v2):
"""
Check if version v1 is less than v2.
Parameters:
- v1 (str): First version
- v2 (str): Second version
Returns:
bool: True if v1 < v2
"""
def version_gte(v1, v2):
"""
Check if version v1 is greater than or equal to v2.
Parameters:
- v1 (str): First version
- v2 (str): Second version
Returns:
bool: True if v1 >= v2
"""Functions for working with Docker environment configuration.
def kwargs_from_env(ssl_version=None, assert_hostname=None, environment=None):
"""
Get Docker client configuration from environment variables.
Parameters:
- ssl_version (str): SSL version for TLS connections
- assert_hostname (bool): Verify hostname in TLS certificates
- environment (dict): Environment variables dict (defaults to os.environ)
Returns:
dict: Client configuration parameters
"""
def datetime_to_timestamp(dt):
"""
Convert datetime object to Unix timestamp.
Parameters:
- dt (datetime): Datetime object
Returns:
int: Unix timestamp
"""
def decode_json_header(header):
"""
Decode base64 JSON header.
Parameters:
- header (str): Base64-encoded JSON header
Returns:
dict: Decoded JSON data
"""Docker registry authentication utilities for managing credentials and login tokens.
def resolve_repository_name(repo_name):
"""
Parse repository name and determine registry.
Parameters:
- repo_name (str): Repository name
Returns:
tuple: (registry_name, repository_name)
"""
def resolve_authconfig(authconfig, registry=None):
"""
Resolve authentication configuration for a registry.
Parameters:
- authconfig (dict): Authentication configurations
- registry (str): Registry URL
Returns:
dict: Authentication configuration for registry
"""
def load_config(config_path=None):
"""
Load Docker configuration file.
Parameters:
- config_path (str): Path to config file (defaults to ~/.docker/config.json)
Returns:
dict: Docker configuration data
"""
def encode_header(auth_config):
"""
Encode authentication configuration for HTTP header.
Parameters:
- auth_config (dict): Authentication configuration
Returns:
str: Base64-encoded authentication header
"""
def get_config_header(client, registry):
"""
Get authentication header for registry from client configuration.
Parameters:
- client: Docker client instance
- registry (str): Registry URL
Returns:
str: Authentication header value
"""Comprehensive TLS/SSL configuration for secure Docker daemon connections.
class TLSConfig:
def __init__(self, client_cert=None, ca_cert=None, verify=None,
ssl_version=None, assert_hostname=None, assert_fingerprint=None):
"""
Create TLS configuration for Docker client.
Parameters:
- client_cert (tuple): (cert_file, key_file) for client authentication
- ca_cert (str): Path to CA certificate file
- verify (bool|str): Verify server certificate (True/False or CA path)
- ssl_version (int): SSL version constant
- assert_hostname (bool): Verify hostname in certificate
- assert_fingerprint (str): Expected certificate fingerprint
"""
def configure_client(self, client):
"""
Configure requests client for TLS connections.
Parameters:
- client: Requests session client
Returns:
None
"""
# Properties
cert: tuple # Client certificate tuple
ca_cert: str # CA certificate path
verify: bool|str # Certificate verification setting
ssl_version: int # SSL version
assert_hostname: bool # Hostname verification
assert_fingerprint: str # Certificate fingerprintSpecialized HTTP adapters for different connection types and platforms.
HTTP adapter for Unix socket connections on Linux/macOS systems.
class UnixAdapter:
def __init__(self, socket_url, timeout=60, num_pools=25):
"""
Create Unix socket HTTP adapter.
Parameters:
- socket_url (str): Unix socket URL
- timeout (int): Connection timeout
- num_pools (int): Connection pool size
"""
class UnixHTTPConnection:
"""HTTP connection over Unix socket"""
class UnixHTTPConnectionPool:
"""Connection pool for Unix socket connections"""HTTP adapter for Windows named pipes.
class NpipeAdapter:
def __init__(self, socket_url, timeout=60, num_pools=25):
"""
Create named pipe HTTP adapter (Windows only).
Parameters:
- socket_url (str): Named pipe URL
- timeout (int): Connection timeout
- num_pools (int): Connection pool size
"""HTTPS adapter with configurable SSL/TLS settings.
class SSLAdapter:
def __init__(self, ssl_version=None, assert_hostname=None,
assert_fingerprint=None, **kwargs):
"""
Create SSL/TLS HTTP adapter.
Parameters:
- ssl_version (int): SSL version to use
- assert_hostname (bool): Verify hostname
- assert_fingerprint (str): Expected certificate fingerprint
"""Decorators for API method enhancement and validation.
def check_resource(f):
"""
Decorator to extract resource ID from function arguments.
Parameters:
- f (function): Function to decorate
Returns:
function: Decorated function
"""
def minimum_version(version):
"""
Decorator to require minimum Docker API version.
Parameters:
- version (str): Minimum required API version
Returns:
function: Decorator function
"""
def update_headers(f):
"""
Decorator to automatically add authentication headers.
Parameters:
- f (function): Function to decorate
Returns:
function: Decorated function
"""Low-level socket operations for Docker API communication.
def read(socket, n=4096):
"""
Read bytes from socket with error handling.
Parameters:
- socket: Socket object
- n (int): Number of bytes to read
Returns:
bytes: Data read from socket
"""
def read_exactly(socket, n):
"""
Read exact number of bytes from socket.
Parameters:
- socket: Socket object
- n (int): Exact number of bytes to read
Returns:
bytes: Exactly n bytes of data
"""
def next_frame_size(socket):
"""
Get size of next frame from multiplexed stream.
Parameters:
- socket: Socket object
Returns:
int: Size of next frame
"""
def frames_iter(socket):
"""
Iterator over frames from multiplexed socket stream.
Parameters:
- socket: Socket object
Yields:
bytes: Frame data
"""
class SocketError(Exception):
"""Exception for socket operation errors"""Functions for handling Docker port specifications and bindings.
def split_port(port):
"""
Parse port specification string.
Parameters:
- port (str): Port specification (e.g., '8080:80/tcp')
Returns:
tuple: (host_port, container_port, protocol)
"""
def build_port_bindings(ports):
"""
Build port bindings dictionary from port list.
Parameters:
- ports (list): List of port specifications
Returns:
dict: Port bindings mapping
"""import docker
from docker.types import LogConfig, Ulimit
client = docker.Client()
# Create comprehensive host configuration
host_config = client.create_host_config(
# Volume bindings
binds={
'/host/data': {'bind': '/container/data', 'mode': 'rw'},
'/host/logs': {'bind': '/container/logs', 'mode': 'rw'}
},
# Port bindings
port_bindings={
'8080/tcp': 8080,
'9000/tcp': ('127.0.0.1', 9000)
},
# Resource limits
mem_limit='512m',
cpu_shares=512,
ulimits=[
Ulimit(name='nofile', soft=1024, hard=2048),
Ulimit(name='nproc', soft=128, hard=256)
],
# Security settings
privileged=False,
cap_add=['NET_ADMIN'],
cap_drop=['MKNOD'],
# Network settings
dns=['8.8.8.8', '8.8.4.4'],
extra_hosts=['api.example.com:192.168.1.100'],
# Logging configuration
log_config=LogConfig(
type=LogConfig.types.JSON,
config={'max-size': '10m', 'max-file': '3'}
),
# Restart policy
restart_policy={'Name': 'unless-stopped'}
)
# Create container with configuration
container = client.create_container(
image='myapp:latest',
command=['python', 'app.py'],
environment={'ENV': 'production'},
host_config=host_config,
labels={'app': 'myservice', 'version': '1.0'}
)from docker.auth import auth
# Load Docker configuration
config = auth.load_config()
print(f"Loaded config with {len(config.get('auths', {}))} registries")
# Resolve authentication for specific registry
auth_config = auth.resolve_authconfig(
config.get('auths', {}),
'registry.example.com'
)
# Use authentication for image operations
client.pull(
'registry.example.com/private/image:latest',
auth_config=auth_config
)
# Manual authentication configuration
manual_auth = {
'username': 'myuser',
'password': 'mypass',
'serveraddress': 'https://registry.example.com'
}
# Encode for header use
auth_header = auth.encode_header(manual_auth)from docker.tls import TLSConfig
import ssl
# Create TLS configuration for secure connection
tls_config = TLSConfig(
client_cert=('/path/to/client-cert.pem', '/path/to/client-key.pem'),
ca_cert='/path/to/ca.pem',
verify=True,
ssl_version=ssl.PROTOCOL_TLSv1_2,
assert_hostname=False
)
# Create client with TLS
client = docker.Client(
base_url='https://docker-daemon:2376',
tls=tls_config
)
# Verify connection
print(client.ping())import os
from docker.utils.utils import kwargs_from_env
# Set environment variables
os.environ['DOCKER_HOST'] = 'tcp://192.168.1.100:2376'
os.environ['DOCKER_TLS_VERIFY'] = '1'
os.environ['DOCKER_CERT_PATH'] = '/path/to/certs'
# Get client configuration from environment
client_kwargs = kwargs_from_env()
print(f"Client config: {client_kwargs}")
# Create client from environment
client = docker.Client(**client_kwargs)
# Alternative: use from_env convenience function
client = docker.from_env()from docker.utils.utils import (
parse_repository_tag, parse_bytes, convert_port_bindings,
convert_volume_binds, split_command
)
# Parse repository and tag
repo, tag = parse_repository_tag('nginx:1.21-alpine')
print(f"Repository: {repo}, Tag: {tag}")
# Parse byte sizes
memory_limit = parse_bytes('512m') # Returns 536870912
cpu_limit = parse_bytes('1g') # Returns 1073741824
# Convert port bindings
port_bindings = convert_port_bindings({
'8080/tcp': 8080,
'9000/tcp': ('0.0.0.0', 9000),
'3306/tcp': None # Random port
})
# Convert volume bindings
volume_binds = convert_volume_binds({
'/host/data': {'bind': '/app/data', 'mode': 'rw'},
'/host/config': {'bind': '/app/config', 'mode': 'ro'}
})
# Split command strings
cmd_array = split_command('python -m http.server 8000')
print(f"Command array: {cmd_array}")from docker.utils.utils import tar, exclude_paths
import tempfile
# Create build context from directory
with tar('/path/to/build/context', gzip=True) as build_context:
# Build image using context
for line in client.build(
fileobj=build_context,
tag='myapp:latest',
custom_context=True,
stream=True,
decode=True
):
if 'stream' in line:
print(line['stream'].strip())
# Handle .dockerignore patterns
root_path = '/path/to/project'
dockerignore_patterns = [
'*.pyc',
'__pycache__/',
'.git/',
'node_modules/',
'*.log'
]
excluded_paths = exclude_paths(root_path, dockerignore_patterns)
print(f"Excluding {len(excluded_paths)} paths from build context")Install with Tessl CLI
npx tessl i tessl/pypi-docker-py