Docker SDK 管理容器与镜像

除 CLI 外，Python 可通过 Docker SDK for Python（docker 包）调用 Docker Engine API，适合批量拉取镜像、巡检容器状态、清理 dangling 镜像、在 CI 中做镜像冒烟测试。

---

环境准备

uv add docker

默认连接本机 /var/run/docker.sock。远程 Docker 需 TLS：

export DOCKER_HOST="tcp://docker-host.example.com:2376"
export DOCKER_TLS_VERIFY="1"
export DOCKER_CERT_PATH="/path/to/certs"  # ca.pem, cert.pem, key.pem

---

连接与列出容器

#!/usr/bin/env python3
"""列出运行中容器，输出名称、镜像、状态、端口映射。"""

import docker


def list_running_containers() -> None:
    client = docker.from_env()  # 读取 DOCKER_HOST 等环境变量

    for c in client.containers.list(filters={"status": "running"}):
        ports = c.attrs.get("NetworkSettings", {}).get("Ports") or {}
        print(f"{c.name:30s}  image={c.image.tags}  ports={ports}")


if __name__ == "__main__":
    list_running_containers()

---

拉取镜像与运行一次性容器

def pull_and_smoke_test(image: str, tag: str = "latest") -> None:
    client = docker.from_env()
    full = f"{image}:{tag}"

    print(f"拉取 {full} ...")
    client.images.pull(image, tag=tag)

    # 运行一次性容器执行健康检查命令
    container = client.containers.run(
        full,
        command=["/bin/sh", "-c", "echo ok && nginx -v"],
        remove=True,       # 退出后自动删除
        detach=False,      # 前台等待结束
    )
    print(f"冒烟测试输出: {container.decode() if isinstance(container, bytes) else container}")


def inspect_image(image_name: str) -> None:
    client = docker.from_env()
    img = client.images.get(image_name)
    print(f"ID: {img.short_id}")
    print(f"Created: {img.attrs['Created']}")
    print(f"Size: {img.attrs['Size'] / 1024 / 1024:.1f} MB")
    for k, v in (img.attrs.get("Config", {}).get("Labels") or {}).items():
        print(f"  label {k}={v}")

---

批量清理与资源巡检

def cleanup_dangling_images(dry_run: bool = True) -> None:
    """删除 dangling 镜像（<none>:<none>）。"""
    client = docker.from_env()
    dangling = client.images.list(filters={"dangling": True})
    for img in dangling:
        print(f"{'[DRY-RUN] ' if dry_run else ''}删除 {img.short_id} size={img.attrs['Size']}")
        if not dry_run:
            client.images.remove(img.id, force=False)


def container_health_report() -> list[dict]:
    """收集所有容器状态，供巡检报表使用。"""
    client = docker.from_env()
    report = []
    for c in client.containers.list(all=True):
        state = c.attrs["State"]
        report.append({
            "name": c.name,
            "status": c.status,
            "running": state["Running"],
            "exit_code": state.get("ExitCode"),
            "started_at": state.get("StartedAt"),
        })
    return report

---

低层 API（等价 docker ps）

需要更细粒度控制时，使用 client.api：

def low_level_ps() -> None:
    client = docker.from_env()
    raw = client.api.containers(all=True, filters={"label": "app=nginx"})
    for item in raw:
        print(item["Names"], item["Status"])

提示

• 批量操作多台主机时，通常 SSH 到各节点执行脚本，或通过 Swarm/K8s 间接管理
• containers.run() 的 volumes、network、environment 参数与 CLI 等价
• 生产清理镜像前确认无容器引用该镜像 ID

---

小结

场景	SDK 方法
连接	docker.from_env()
列容器	client.containers.list()
拉镜像	client.images.pull()
跑容器	client.containers.run()
删镜像	client.images.remove()

下一章介绍 Kubernetes API，在集群层面管理 Pod 与 Deployment。