Stage 7 · Master
Kubernetes & Container Automation
Docker SDK for Python
Build, run, and inspect containers programmatically for testing pipelines.
Docker SDK Setup
The Docker SDK for Python talks to the Docker daemon via the Unix socket or TCP. It supports container, image, network, and volume operations.
import docker
# Connect to local Docker daemon
client = docker.from_env()
# Verify connection
print(f"Docker version: {client.version()['Version']}")
print(f"Containers: {len(client.containers.list())}")
print(f"Images: {len(client.images.list())}")
# Connect to remote Docker daemon
client = docker.DockerClient(base_url="tcp://localhost:2375")docker.from_env() reads DOCKER_HOST, DOCKER_CERT_PATH, and DOCKER_TLS_VERIFY from environment variables. For remote daemons, pass base_url explicitly.
Container Operations
import docker
client = docker.from_env()
# Run a container
container = client.containers.run(
image="nginx:latest",
name="my-nginx",
ports={"80/tcp": 8080},
detach=True, # Run in background
environment={"NGINX_HOST": "localhost"},
)
print(f"Container ID: {container.short_id}")
# List running containers
for c in client.containers.list():
print(f"{c.name}: {c.status}")
# List all containers (including stopped)
for c in client.containers.list(all=True):
print(f"{c.name}: {c.status}")
# Get container logs
logs = container.logs(stream=True, tail=100)
for line in logs:
print(line.decode().rstrip())
# Execute a command in a running container
exit_code, output = container.exec_run("df -h")
print(output.decode())
# Stop and remove
container.stop()
container.remove()detach=True returns the container object without waiting for it to finish. Without detach, run() blocks until the container exits. Use detach for long-running containers.
Image Operations
import docker
client = docker.from_env()
# List images
for image in client.images.list():
tags = ", ".join(image.tags) if image.tags else "<none>"
size_mb = image.attrs["Size"] / (1024 * 1024)
print(f"{tags}: {size_mb:.1f} MB")
# Pull an image
image = client.images.pull("python", tag="3.12-slim")
print(f"Pulled: {image.tags}")
# Inspect image layers
for layer in image.history():
print(f"{layer['CreatedBy'][:80]}: {layer['Size']} bytes")
# Remove an image
client.images.remove("python:3.12-slim")
# Remove all unused images
client.images.prune()image.history() shows the layers that make up an image. Each layer is a read-only filesystem snapshot. Understanding layers helps you write efficient Dockerfiles.
Building Images
import docker
from pathlib import Path
client = docker.from_env()
# Build an image from a Dockerfile
image, build_logs = client.images.build(
path=".",
dockerfile="Dockerfile",
tag="myapp:latest",
rm=True, # Remove intermediate containers
)
for chunk in build_logs:
if "stream" in chunk:
print(chunk["stream"].rstrip())
# Build with build arguments
image, _ = client.images.build(
path=".",
tag="myapp:latest",
buildargs={"PYTHON_VERSION": "3.12"},
)
# Build from a fileobj (Dockerfile as string)
dockerfile = """
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "main.py"]
"""
import io
fileobj = io.BytesIO(dockerfile.encode())
image, _ = client.images.build(fileobj=fileobj, tag="myapp:inline")build() returns the image object and build logs. rm=True removes intermediate containers to keep the image small. buildargs pass values to ARG instructions in the Dockerfile.
Network and Volume Management
import docker
client = docker.from_env()
# Create a network
network = client.networks.create("my-network", driver="bridge")
print(f"Network: {network.name}")
# Connect a container to the network
container = client.containers.run("nginx:latest", detach=True)
network.connect(container)
# List networks
for net in client.networks.list():
print(f"{net.name}: {net.attrs['Driver']}")
# Create a volume
volume = client.volumes.create("my-data", driver="local")
print(f"Volume: {volume.name}")
# Run a container with the volume
container = client.containers.run(
"postgres:16",
detach=True,
volumes={"my-data": {"bind": "/var/lib/postgresql/data", "mode": "rw"}},
environment={"POSTGRES_PASSWORD": "secret"},
)
# Clean up
container.stop()
container.remove()
volume.remove()
network.remove()Networks let containers communicate by name. Volumes persist data beyond container lifetime. Always clean up test resources to avoid disk and network sprawl.
Docker Compose from Python
import subprocess
from pathlib import Path
def docker_compose(command: str, project_dir: str = ".", **kwargs) -> str:
"""Run a docker compose command."""
cmd = ["docker", "compose", command]
for key, value in kwargs.items():
cmd.extend([f"--{key.replace('_', '-')}", str(value)])
result = subprocess.run(cmd, capture_output=True, text=True, cwd=project_dir)
if result.returncode != 0:
raise Exception(f"docker compose failed: {result.stderr}")
return result.stdout
# Start services
output = docker_compose("up", detach=True)
print(output)
# List running services
output = docker_compose("ps")
print(output)
# View logs
output = docker_compose("logs", tail="100", services="web")
print(output)
# Stop and remove
docker_compose("down", volumes=True)The Docker Compose CLI can be invoked from Python using subprocess. This is simpler than the Python SDK for compose operations. Use the SDK for individual container management.
The Docker SDK is ideal for integration test pipelines. Spin up test databases, message queues, and mock services programmatically. Clean them up after tests finish.
Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.