Python Docker SDK 全面教程:常用 API 串联与实战指南
·
大家好,我是jobleap.cn的小九。
如果你想了解 Python 的 Docker 组件(Docker SDK for Python)的核心用途、解决的问题,并通过系统的教程掌握其常用 API 的使用方法,我会从基础到实战,帮你全面理解这个工具。
一、Python Docker SDK 核心价值:能做什么?解决什么问题?
1. 核心用途
Python Docker SDK 是 Docker 官方提供的 Python 编程接口,让你可以通过 Python 代码而非命令行来操作 Docker 引擎,实现对容器、镜像、网络、卷等 Docker 核心资源的全生命周期管理。
2. 解决的核心问题
- 替代手动命令行操作:避免在自动化脚本中调用
subprocess执行docker命令,代码更易维护、可读性更高; - 自动化容器编排:批量创建/启动/停止/删除容器、镜像,适合测试环境搭建、服务批量部署等场景;
- 集成到业务系统:将 Docker 操作嵌入 Python 应用(如运维平台、CI/CD 工具),实现容器化服务的动态管理;
- 精细化控制 Docker 资源:通过 API 精确配置容器的网络、存储、环境变量等参数,比命令行更灵活。
二、环境准备
1. 安装依赖
首先确保本地已安装 Docker 引擎(启动状态),然后安装 Python Docker SDK:
pip install docker
2. 连接验证
Docker SDK 默认连接本地 Docker 引擎(unix://var/run/docker.sock),远程连接需配置环境变量或指定参数。先验证连接是否正常:
import docker
# 创建 Docker 客户端
client = docker.from_env()
# 验证连接:打印 Docker 服务器版本
try:
version_info = client.version()
print(f"Docker 服务器版本:{version_info['Version']}")
print("✅ 连接 Docker 引擎成功!")
except Exception as e:
print(f"❌ 连接失败:{e}")
输出示例:
Docker 服务器版本:24.0.7
✅ 连接 Docker 引擎成功!
三、常用 API 实战:核心功能串联
1. 镜像管理(拉取、查看、删除)
import docker
client = docker.from_env()
# 1. 拉取镜像(以 nginx:alpine 为例)
print("📥 拉取 nginx:alpine 镜像...")
image = client.images.pull("nginx:alpine")
print(f"拉取成功,镜像 ID:{image.id[:12]}")
# 2. 查看本地所有镜像
print("\n📋 本地镜像列表:")
for img in client.images.list():
# 提取镜像标签(无标签则显示 <none>)
tags = img.tags[0] if img.tags else "<none>"
print(f"- {tags} (ID: {img.id[:12]})")
# 3. 删除镜像(可选,注意:被容器使用的镜像无法直接删除)
# client.images.remove("nginx:alpine")
# print("删除 nginx:alpine 镜像成功")
2. 容器管理(创建、启动、查看、停止、删除)
这是 Docker SDK 最核心的功能,我们串联“创建→启动→查看状态→执行命令→停止→删除”全流程:
import docker
import time
client = docker.from_env()
# 1. 创建并启动容器(等价于 docker run -d -p 8080:80 --name my-nginx nginx:alpine)
print("🚀 创建并启动容器...")
container = client.containers.run(
image="nginx:alpine",
name="my-nginx",
ports={"80/tcp": 8080}, # 端口映射:主机 8080 → 容器 80
detach=True, # 后台运行
restart_policy={"Name": "on-failure", "MaximumRetryCount": 3} # 重启策略
)
print(f"容器启动成功,容器 ID:{container.id[:12]}")
# 2. 查看容器状态
time.sleep(1) # 等待容器完全启动
container.reload() # 刷新容器状态
print(f"\n📊 容器状态:{container.status}")
print(f"容器端口映射:{container.attrs['NetworkSettings']['Ports']}")
# 3. 在容器内执行命令(等价于 docker exec my-nginx ls /usr/share/nginx/html)
print("\n💻 执行容器内命令:ls /usr/share/nginx/html")
exec_result = container.exec_run("ls /usr/share/nginx/html")
print(f"命令输出:{exec_result.output.decode('utf-8').strip()}")
# 4. 查看容器日志(等价于 docker logs my-nginx)
print("\n📄 容器最新日志:")
logs = container.logs(tail=5) # 只看最后 5 行
print(logs.decode('utf-8'))
# 5. 停止容器
print("\n🛑 停止容器...")
container.stop()
container.reload()
print(f"容器状态:{container.status}")
# 6. 删除容器
print("\n🗑️ 删除容器...")
container.remove()
# 验证删除:查看所有运行/停止的容器,确认 my-nginx 不存在
all_containers = [c.name for c in client.containers.list(all=True)]
print(f"容器是否已删除:{'my-nginx' not in all_containers}")
3. 网络与卷管理(进阶)
针对“容器网络隔离”和“数据持久化”需求,演示网络创建、卷挂载:
import docker
client = docker.from_env()
# 1. 创建自定义网络(等价于 docker network create my-network)
print("🌐 创建自定义网络 my-network...")
network = client.networks.create(
name="my-network",
driver="bridge", # 网络驱动(默认 bridge)
ipam={"Driver": "default", "Config": [{"Subnet": "172.20.0.0/16"}]}
)
print(f"网络创建成功,网络 ID:{network.id[:12]}")
# 2. 创建数据卷(等价于 docker volume create my-nginx-volume)
print("\n📦 创建数据卷 my-nginx-volume...")
volume = client.volumes.create(name="my-nginx-volume")
print(f"卷创建成功,卷名称:{volume.name}")
# 3. 启动容器并挂载卷、加入自定义网络
print("\n🚀 启动容器并关联网络/卷...")
container = client.containers.run(
image="nginx:alpine",
name="my-nginx-with-volume",
ports={"80/tcp": 8081},
volumes={volume.name: {"bind": "/usr/share/nginx/html", "mode": "rw"}}, # 卷挂载
networks=["my-network"], # 加入自定义网络
detach=True
)
print(f"容器启动成功,挂载卷:{container.attrs['Mounts'][0]['Name']}")
# 清理资源(可选)
container.stop()
container.remove()
client.networks.get("my-network").remove()
client.volumes.get("my-nginx-volume").remove()
print("\n🧹 清理网络/卷/容器完成")
四、实战场景:批量管理容器
模拟“批量启动 3 个 Nginx 容器 + 统一监控状态 + 批量停止删除”的运维场景:
import docker
client = docker.from_env()
# 批量启动容器
container_names = ["nginx-1", "nginx-2", "nginx-3"]
started_containers = []
print("📦 批量启动 Nginx 容器...")
for i, name in enumerate(container_names):
port = 8081 + i # 端口依次为 8081、8082、8083
container = client.containers.run(
image="nginx:alpine",
name=name,
ports={"80/tcp": port},
detach=True
)
started_containers.append(container)
print(f"- {name} (端口 {port}) 启动成功,ID:{container.id[:12]}")
# 批量检查状态
print("\n🔍 批量检查容器状态:")
for container in started_containers:
container.reload()
print(f"- {container.name}: {container.status}")
# 批量停止并删除
print("\n🧹 批量停止并删除容器...")
for container in started_containers:
container.stop()
container.remove()
print(f"- {container.name} 已停止并删除")
五、异常处理(生产环境必备)
实际开发中需捕获 Docker 相关异常,避免程序崩溃:
import docker
from docker.errors import DockerException, ImageNotFound, ContainerAlreadyExists
client = docker.from_env()
try:
# 尝试创建已存在的容器(触发异常)
container = client.containers.run(
image="nginx:alpine",
name="my-nginx",
detach=True
)
except ContainerAlreadyExists:
print("❌ 容器已存在,请更换名称!")
except ImageNotFound:
print("❌ 镜像不存在,请先拉取!")
except DockerException as e:
print(f"❌ Docker 操作失败:{str(e)}")
finally:
# 清理残留容器
try:
client.containers.get("my-nginx").remove(force=True)
except:
pass
总结
- 核心定位:Python Docker SDK 是操作 Docker 引擎的编程接口,替代命令行实现容器/镜像/网络/卷的自动化管理,解决手动操作效率低、难以集成到业务系统的问题;
- 核心 API 流程:镜像(pull/list/remove)→ 容器(run/exec/stop/remove)→ 网络/卷(create/mount)是最常用的串联逻辑;
- 实战要点:生产环境中需添加异常处理、状态刷新(
reload())、资源清理,确保操作的稳定性和可维护性。
通过这套教程,你可以基于 Python 快速实现 Docker 资源的自动化管理,无论是搭建测试环境、开发运维工具,还是集成到 CI/CD 流程,都能大幅提升效率。
更多推荐


所有评论(0)