Python Docker库完全指南：用代码掌控容器生命周期

一、Python生态与Docker库简介

Python凭借简洁的语法、丰富的生态系统和强大的扩展性，已成为全球最受欢迎的编程语言之一。它在Web开发、数据分析、人工智能、自动化运维、云计算等众多领域都发挥着不可替代的作用。在DevOps和云原生技术快速发展的今天，容器化技术成为连接开发与运维的关键桥梁，而Docker作为容器化领域的事实标准，其重要性不言而喻。本文将聚焦Python生态中操作Docker的核心库——docker，通过实例详解如何用Python代码实现容器的全生命周期管理，让开发者无需手动输入命令即可掌控Docker容器。

二、Docker库核心解析

2.1 库的用途

docker库是Docker官方提供的Python SDK，它允许开发者通过Python代码与Docker引擎进行交互，实现容器创建、启动、停止、删除，镜像构建、推送、拉取等所有Docker命令行工具能完成的操作。这为自动化部署、持续集成/持续部署(CI/CD)流程、容器编排等场景提供了强大的编程接口。

2.2 工作原理

该库通过封装Docker Engine API，使Python代码能够通过HTTP请求与本地或远程的Docker守护进程通信。它采用客户端-服务器架构，客户端负责发送指令，Docker引擎负责实际执行容器和镜像的管理操作。库内部使用requests等HTTP库处理API调用，将复杂的RESTful API交互抽象为简洁的Python方法。

2.3 优缺点分析

优点：

• 官方维护，与Docker引擎兼容性强
• 接口设计直观，贴近Docker CLI命令
• 支持所有Docker核心功能
• 详细的错误处理和日志输出

缺点：

• 高级功能需要深入理解Docker内部原理
• 异步操作支持不够完善
• 某些复杂场景下配置选项较多，学习曲线较陡

2.4 许可证类型

docker库采用Apache License 2.0开源许可证，允许商业使用、修改、分发和私人使用，但要求在修改后的代码中保留原始版权声明和许可条款。

三、Docker库安装与环境配置

3.1 安装前提

在安装docker库之前，需要确保系统满足以下条件：

• 已安装Python 3.6或更高版本
• 已安装Docker Engine（社区版或企业版）
• Docker守护进程正在运行
• 对于Linux系统，当前用户已加入docker用户组（避免使用sudo）

3.2 安装步骤

通过pip工具可以轻松安装docker库：

# 安装最新稳定版
pip install docker

# 安装指定版本
pip install docker==6.1.3

# 安装包含所有可选依赖的完整版
pip install "docker[ssh,tls]"

安装完成后，可以通过以下命令验证安装是否成功：

python -c "import docker; print(f'Docker SDK for Python version: {docker.__version__}')"

如果输出类似Docker SDK for Python version: 6.1.3的信息，则表示安装成功。

3.3 环境配置

本地Docker连接配置

默认情况下，docker库会通过以下方式查找Docker守护进程：

• Unix系统：unix://var/run/docker.sock
• Windows系统：npipe:////./pipe/docker_engine

如果你的Docker守护进程使用非默认配置，可以通过环境变量指定连接地址：

# Linux/macOS
export DOCKER_HOST=tcp://127.0.0.1:2375

# Windows (PowerShell)
$env:DOCKER_HOST = "tcp://127.0.0.1:2375"

远程Docker连接配置

连接远程Docker守护进程时，需要配置TLS验证（推荐）或允许非加密连接（不推荐用于生产环境）：

import docker

# 远程连接（无TLS验证，不安全）
client = docker.DockerClient(base_url='tcp://remote-docker-host:2375')

# 远程连接（带TLS验证）
client = docker.DockerClient(
    base_url='tcp://remote-docker-host:2376',
    tls=True,
    tls_verify=True,
    ca_cert='/path/to/ca.pem',
    cert='/path/to/cert.pem',
    key='/path/to/key.pem'
)

四、Docker库核心功能详解

4.1 客户端初始化

所有Docker操作都始于客户端对象的创建，它是与Docker引擎交互的入口：

import docker

# 创建默认客户端
client = docker.from_env()

# 验证连接是否成功
try:
    client.ping()
    print("成功连接到Docker守护进程")
except docker.errors.APIError as e:
    print(f"连接Docker失败: {e}")

docker.from_env()方法会自动从环境变量中读取Docker配置，包括DOCKER_HOST、DOCKER_TLS_VERIFY和DOCKER_CERT_PATH等。

4.2 镜像管理

4.2.1 镜像拉取

从Docker仓库拉取镜像到本地：

import docker
from docker.errors import APIError

client = docker.from_env()

def pull_image(image_name, tag='latest'):
    """拉取Docker镜像"""
    try:
        print(f"开始拉取镜像: {image_name}:{tag}")
        image = client.images.pull(image_name, tag=tag)
        print(f"镜像拉取成功: {image.tags[0]}")
        return image
    except APIError as e:
        print(f"镜像拉取失败: {e}")
        return None

# 拉取官方Python镜像
pull_image('python', '3.10-slim')

# 拉取私有仓库镜像（需要先登录）
client.login(username='your_username', password='your_password', registry='your.registry.com')
pull_image('your.registry.com/your-project/app', 'v1.0')

4.2.2 镜像列表查询

查看本地已有的Docker镜像：

def list_images(filter_str=None):
    """列出本地Docker镜像"""
    images = client.images.list()
    if filter_str:
        images = [img for img in images if any(filter_str in tag for tag in img.tags)]

    print(f"找到 {len(images)} 个镜像:")
    for img in images:
        tags = img.tags if img.tags else ['<none>:<none>']
        print(f"ID: {img.id[:12]}, Tags: {', '.join(tags)}, Size: {img.attrs['Size']//(1024*1024)}MB")

# 列出所有镜像
list_images()

# 列出包含python的镜像
list_images('python')

4.2.3 镜像构建

从Dockerfile构建自定义镜像：

def build_image(dockerfile_path, image_name, tag='latest', build_args=None):
    """从Dockerfile构建镜像"""
    try:
        print(f"开始构建镜像: {image_name}:{tag}")
        # 构建参数
        buildargs = build_args if build_args else {}

        # 构建镜像
        image, build_logs = client.images.build(
            path=dockerfile_path,
            tag=f"{image_name}:{tag}",
            buildargs=buildargs,
            rm=True  # 构建完成后删除中间容器
        )

        # 输出构建日志
        for log in build_logs:
            if 'stream' in log:
                print(log['stream'].strip())

        print(f"镜像构建成功: {image.tags[0]}")
        return image
    except APIError as e:
        print(f"镜像构建失败: {e}")
        return None

# 构建示例：从当前目录的Dockerfile构建镜像
build_image(
    dockerfile_path='.',
    image_name='my-python-app',
    tag='v1.0',
    build_args={'PYTHON_VERSION': '3.10'}
)

假设当前目录有以下Dockerfile：

ARG PYTHON_VERSION
FROM python:${PYTHON_VERSION}-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

CMD ["python", "app.py"]

4.2.4 镜像推送与删除

将本地镜像推送到远程仓库并删除不需要的镜像：

def push_and_cleanup(image_name, tag='latest'):
    """推送镜像到仓库并清理本地镜像"""
    try:
        # 推送镜像
        print(f"推送镜像到仓库: {image_name}:{tag}")
        push_logs = client.images.push(image_name, tag=tag)
        print(push_logs)

        # 删除本地镜像
        print(f"删除本地镜像: {image_name}:{tag}")
        client.images.remove(image=f"{image_name}:{tag}")
        print("操作完成")
    except APIError as e:
        print(f"操作失败: {e}")

# 推送并清理镜像
push_and_cleanup('your.registry.com/your-project/app', 'v1.0')

4.3 容器管理

4.3.1 创建与启动容器

创建并启动一个Docker容器：

def create_and_start_container(image_name, container_name, command=None, ports=None, volumes=None, environment=None):
    """创建并启动容器"""
    try:
        # 端口映射格式: {'容器端口/tcp': 主机端口}
        port_bindings = ports if ports else {}

        # 数据卷映射格式: {'主机路径': {'bind': '容器路径', 'mode': 'ro'}}
        volume_mounts = volumes if volumes else {}

        # 环境变量格式: {'KEY': 'VALUE'}
        env_vars = environment if environment else {}

        print(f"创建容器: {container_name} 使用镜像: {image_name}")
        container = client.containers.create(
            image=image_name,
            name=container_name,
            command=command,
            ports=port_bindings,
            volumes=volume_mounts,
            environment=env_vars,
            detach=True  # 后台运行
        )

        print(f"启动容器: {container_name}")
        container.start()
        print(f"容器 {container_name} 启动成功，ID: {container.id[:12]}")
        return container
    except APIError as e:
        print(f"容器操作失败: {e}")
        return None

# 创建并启动一个Python应用容器
create_and_start_container(
    image_name='my-python-app:v1.0',
    container_name='python-app-container',
    ports={'5000/tcp': 5000},  # 容器5000端口映射到主机5000端口
    volumes={'/host/path/data': {'bind': '/app/data', 'mode': 'rw'}},  # 数据卷映射
    environment={'FLASK_ENV': 'production', 'PORT': '5000'},  # 环境变量
    command='python app.py'  # 启动命令
)

4.3.2 容器状态管理

查看容器状态、停止、启动和重启容器：

def manage_container(container_name, action='status'):
    """管理容器状态"""
    try:
        # 获取容器对象
        container = client.containers.get(container_name)

        if action == 'status':
            print(f"容器 {container_name} 状态:")
            print(f"ID: {container.id[:12]}")
            print(f"状态: {container.status}")
            print(f"镜像: {container.image.tags[0]}")
            print(f"启动时间: {container.attrs['Created']}")
            print(f"端口映射: {container.attrs['HostConfig']['PortBindings']}")
            print(f"IP地址: {container.attrs['NetworkSettings']['IPAddress']}")
            return container.status

        elif action == 'stop':
            print(f"停止容器: {container_name}")
            container.stop()
            print(f"容器 {container_name} 已停止")

        elif action == 'start':
            print(f"启动容器: {container_name}")
            container.start()
            print(f"容器 {container_name} 已启动")

        elif action == 'restart':
            print(f"重启容器: {container_name}")
            container.restart()
            print(f"容器 {container_name} 已重启")

        elif action == 'logs':
            print(f"容器 {container_name} 日志:")
            print(container.logs(tail=50).decode('utf-8'))  # 打印最后50行日志

        else:
            print(f"不支持的操作: {action}")

    except APIError as e:
        print(f"容器管理失败: {e}")

# 查看容器状态
manage_container('python-app-container', 'status')

# 查看容器日志
manage_container('python-app-container', 'logs')

# 重启容器
manage_container('python-app-container', 'restart')

4.3.3 容器列表查询

列出所有容器（包括运行中和已停止的）：

def list_containers(all_containers=True, filter_str=None):
    """列出容器"""
    containers = client.containers.list(all=all_containers)

    if filter_str:
        containers = [c for c in containers if filter_str in c.name]

    print(f"找到 {len(containers)} 个容器:")
    for container in containers:
        status = container.status
        image = container.image.tags[0] if container.image.tags else '<none>'
        ports = []
        if container.attrs['NetworkSettings']['Ports']:
            for port in container.attrs['NetworkSettings']['Ports']:
                if container.attrs['NetworkSettings']['Ports'][port]:
                    ports.append(f"{container.attrs['NetworkSettings']['Ports'][port][0]['HostPort']}->{port.split('/')[0]}")
        ports_str = ', '.join(ports) if ports else '无'

        print(f"名称: {container.name}, ID: {container.id[:12]}, 状态: {status}, 镜像: {image}, 端口: {ports_str}")

# 列出所有容器
list_containers()

# 列出包含python的运行中容器
list_containers(all_containers=False, filter_str='python')

4.3.4 容器删除与清理

删除容器和清理无用容器：

def remove_container(container_name, force=False, volumes=False):
    """删除容器"""
    try:
        container = client.containers.get(container_name)

        # 如果容器正在运行且force=True，则先停止容器
        if container.status == 'running' and force:
            print(f"强制停止容器: {container_name}")
            container.stop()

        print(f"删除容器: {container_name}")
        container.remove(v volumes=volumes)  # volumes=True表示同时删除关联的数据卷
        print(f"容器 {container_name} 已删除")

    except APIError as e:
        print(f"删除容器失败: {e}")

def cleanup_containers():
    """清理所有已停止的容器"""
    try:
        stopped_containers = client.containers.list(all=True, filters={'status': 'exited'})
        if not stopped_containers:
            print("没有已停止的容器需要清理")
            return

        print(f"找到 {len(stopped_containers)} 个已停止的容器，开始清理...")
        for container in stopped_containers:
            print(f"删除容器: {container.name}")
            container.remove()
        print("清理完成")

    except APIError as e:
        print(f"清理容器失败: {e}")

# 删除指定容器
remove_container('python-app-container', force=True, volumes=True)

# 清理所有已停止的容器
cleanup_containers()

4.4 网络管理

Docker网络允许容器之间通信和容器与外部网络通信，docker库提供了完整的网络管理功能：

4.4.1 创建和管理网络

def list_networks(filter_str=None):
    """列出所有网络"""
    networks = client.networks.list()

    if filter_str:
        networks = [net for net in networks if filter_str in net.name]

    print(f"找到 {len(networks)} 个网络:")
    for network in networks:
        print(f"名称: {network.name}, ID: {network.id[:12]}, 驱动: {network.attrs['Driver']}")
        if 'IPAM' in network.attrs and 'Config' in network.attrs['IPAM']:
            for config in network.attrs['IPAM']['Config']:
                subnet = config.get('Subnet', 'N/A')
                gateway = config.get('Gateway', 'N/A')
                print(f"  子网: {subnet}, 网关: {gateway}")

# 创建自定义网络
create_network(
    network_name='my-app-network',
    subnet='172.18.0.0/16',
    gateway='172.18.0.1'
)

# 列出所有网络
list_networks()

4.4.2 容器连接网络

将容器连接到指定网络：

def connect_container_to_network(container_name, network_name):
    """将容器连接到网络"""
    try:
        # 获取容器和网络对象
        container = client.containers.get(container_name)
        network = client.networks.get(network_name)

        # 检查容器是否已连接到网络
        for net in container.attrs['NetworkSettings']['Networks']:
            if net == network_name:
                print(f"容器 {container_name} 已连接到网络 {network_name}")
                return

        print(f"将容器 {container_name} 连接到网络 {network_name}")
        network.connect(container)
        print(f"连接成功")

    except APIError as e:
        print(f"网络连接失败: {e}")

# 创建容器并连接到网络
create_and_start_container(
    image_name='nginx:alpine',
    container_name='web-server',
    ports={'80/tcp': 8080}
)

# 连接到自定义网络
connect_container_to_network('web-server', 'my-app-network')

4.5 数据卷管理

数据卷是Docker中持久化数据的推荐方式，docker库提供了完整的数据卷管理功能：

4.5.1 创建和管理数据卷

def create_volume(volume_name, driver='local', driver_opts=None):
    """创建数据卷"""
    try:
        # 检查数据卷是否已存在
        existing_volumes = client.volumes.list(filters={'name': volume_name})
        if existing_volumes:
            print(f"数据卷 {volume_name} 已存在")
            return existing_volumes[0]

        print(f"创建数据卷: {volume_name}")
        volume = client.volumes.create(
            name=volume_name,
            driver=driver,
            driver_opts=driver_opts if driver_opts else {}
        )
        print(f"数据卷 {volume_name} 创建成功，ID: {volume.id[:12]}")
        return volume
    except APIError as e:
        print(f"数据卷创建失败: {e}")
        return None

def list_volumes(filter_str=None):
    """列出所有数据卷"""
    volumes = client.volumes.list()

    if filter_str:
        volumes = [vol for vol in volumes if filter_str in vol.name]

    print(f"找到 {len(volumes)} 个数据卷:")
    for volume in volumes:
        print(f"名称: {volume.name}, ID: {volume.id[:12]}, 驱动: {volume.attrs['Driver']}")
        print(f"  挂载点: {volume.attrs['Mountpoint']}")

# 创建数据卷
create_volume('my-data-volume')

# 列出所有数据卷
list_volumes()

4.5.2 使用数据卷启动容器

# 创建数据卷
data_volume = create_volume('app-data')

# 使用数据卷启动容器
create_and_start_container(
    image_name='postgres:14-alpine',
    container_name='postgres-db',
    ports={'5432/tcp': 5432},
    volumes={data_volume.name: {'bind': '/var/lib/postgresql/data', 'mode': 'rw'}},
    environment={
        'POSTGRES_USER': 'admin',
        'POSTGRES_PASSWORD': 'password',
        'POSTGRES_DB': 'mydatabase'
    }
)

4.6 Docker Compose集成

虽然docker库本身不直接支持Docker Compose，但可以通过docker-compose命令行工具的API实现集成：

import subprocess
import os

def run_docker_compose(compose_file, action='up', options='-d'):
    """运行Docker Compose命令"""
    try:
        # 构建命令
        cmd = ['docker-compose', '-f', compose_file, action]
        if options:
            cmd.extend(options.split())

        print(f"执行命令: {' '.join(cmd)}")
        result = subprocess.run(
            cmd,
            cwd=os.path.dirname(os.path.abspath(compose_file)),
            capture_output=True,
            text=True
        )

        if result.returncode != 0:
            print(f"命令执行失败: {result.stderr}")
            return False

        print(f"命令执行成功: {result.stdout}")
        return True
    except Exception as e:
        print(f"执行Docker Compose命令失败: {e}")
        return False

# 示例：使用Docker Compose启动应用
compose_file = 'docker-compose.yml'

# 写入示例Docker Compose文件
with open(compose_file, 'w') as f:
    f.write("""
version: '3'
services:
  web:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
  app:
    image: python:3.10-slim
    command: python -m http.server 8000
    volumes:
      - .:/app
    working_dir: /app
""")

# 启动服务
run_docker_compose(compose_file, 'up', '-d')

# 停止服务
# run_docker_compose(compose_file, 'down')

五、Docker库在实际项目中的应用

5.1 自动化测试环境部署

在CI/CD流程中，使用docker库自动部署测试环境：

import docker
import time

def setup_test_environment():
    """设置测试环境"""
    client = docker.from_env()

    try:
        # 清理旧环境
        print("清理旧的测试环境...")
        for container in client.containers.list(all=True):
            if 'test-' in container.name:
                container.remove(force=True)

        # 创建网络
        print("创建测试网络...")
        network = client.networks.create('test-network', driver='bridge')

        # 启动数据库服务
        print("启动数据库服务...")
        db_container = client.containers.run(
            image='postgres:14-alpine',
            name='test-db',
            environment={
                'POSTGRES_USER': 'test',
                'POSTGRES_PASSWORD': 'test',
                'POSTGRES_DB': 'testdb'
            },
            networks=[network.name],
            detach=True
        )

        # 等待数据库启动
        print("等待数据库服务启动...")
        time.sleep(10)

        # 启动应用服务
        print("启动应用服务...")
        app_container = client.containers.run(
            image='my-app:test',
            name='test-app',
            ports={'8000/tcp': 8000},
            environment={
                'DB_HOST': 'test-db',
                'DB_USER': 'test',
                'DB_PASSWORD': 'test',
                'DB_NAME': 'testdb'
            },
            networks=[network.name],
            detach=True
        )

        print("测试环境设置完成！")
        return {
            'network': network,
            'db_container': db_container,
            'app_container': app_container
        }

    except Exception as e:
        print(f"设置测试环境失败: {e}")
        return None

# 使用示例
test_env = setup_test_environment()
if test_env:
    # 运行测试
    print("运行测试...")
    # 这里可以执行测试命令

    # 清理环境
    print("清理测试环境...")
    test_env['app_container'].remove(force=True)
    test_env['db_container'].remove(force=True)
    test_env['network'].remove()

5.2 微服务动态扩展

根据负载情况动态扩展微服务实例：

import docker
import time
from prometheus_client import CollectorRegistry, Counter, push_to_gateway

def scale_service(service_name, target_count):
    """扩展或收缩服务实例"""
    client = docker.from_env()

    try:
        # 获取当前运行的服务实例
        running_containers = client.containers.list(
            filters={'name': service_name}
        )

        current_count = len(running_containers)
        print(f"当前 {service_name} 实例数: {current_count}")
        print(f"目标 {service_name} 实例数: {target_count}")

        # 扩展服务
        if target_count > current_count:
            print(f"需要扩展 {service_name} 服务，增加 {target_count - current_count} 个实例")
            image = running_containers[0].image.tags[0] if running_containers else f'{service_name}:latest'

            for i in range(current_count, target_count):
                container_name = f"{service_name}-{i+1}"
                print(f"创建实例: {container_name}")

                # 获取原始容器的配置
                if running_containers:
                    config = running_containers[0].attrs
                    ports = config['HostConfig']['PortBindings']
                    env = config['Config']['Env']
                    volumes = config['HostConfig']['Binds']
                else:
                    ports = None
                    env = None
                    volumes = None

                # 创建新容器
                client.containers.run(
                    image=image,
                    name=container_name,
                    ports=ports,
                    environment=env,
                    volumes=volumes,
                    detach=True
                )

        # 收缩服务
        elif target_count < current_count:
            print(f"需要收缩 {service_name} 服务，减少 {current_count - target_count} 个实例")
            containers_to_remove = running_containers[target_count:]

            for container in containers_to_remove:
                print(f"移除实例: {container.name}")
                container.remove(force=True)

        print(f"{service_name} 服务扩展完成，当前实例数: {target_count}")

        # 记录扩展操作
        registry = CollectorRegistry()
        c = Counter('service_scaling', 'Number of service scaling operations', ['service', 'direction'], registry=registry)
        if target_count > current_count:
            c.labels(service=service_name, direction='up').inc(target_count - current_count)
        elif target_count < current_count:
            c.labels(service=service_name, direction='down').inc(current_count - target_count)
        push_to_gateway('prometheus-pushgateway:9091', job='service_scaler', registry=registry)

        return True

    except Exception as e:
        print(f"服务扩展失败: {e}")
        return False

# 基于负载的自动扩展示例
def auto_scale_based_on_load(service_name, min_instances=1, max_instances=5, threshold=70):
    """基于负载的自动扩展"""
    while True:
        # 获取当前负载（这里简化为随机数）
        current_load = get_current_load()  # 实际项目中应该从监控系统获取

        # 计算需要的实例数
        if current_load > threshold:
            current_instances = len(client.containers.list(filters={'name': service_name}))
            target_instances = min(max_instances, current_instances + 1)
            if target_instances > current_instances:
                print(f"高负载检测: {current_load}%，扩展服务到 {target_instances} 个实例")
                scale_service(service_name, target_instances)
        elif current_load < threshold * 0.5:
            current_instances = len(client.containers.list(filters={'name': service_name}))
            target_instances = max(min_instances, current_instances - 1)
            if target_instances < current_instances:
                print(f"低负载检测: {current_load}%，收缩服务到 {target_instances} 个实例")
                scale_service(service_name, target_instances)

        # 等待一段时间再检查
        time.sleep(60)  # 每分钟检查一次

# 模拟获取负载
def get_current_load():
    import random
    return random.randint(20, 90)

# 使用示例
scale_service('web-api', 3)  # 扩展到3个实例
# auto_scale_based_on_load('web-api')  # 启动自动扩展

5.3 自定义镜像构建流水线

构建、测试和推送Docker镜像的自动化流水线：

import docker
import subprocess
import os
import time
from datetime import datetime

def build_and_push_image(repo_path, image_name, tags=None):
    """构建、测试并推送Docker镜像"""
    client = docker.from_env()

    try:
        # 默认标签使用当前时间戳
        if not tags:
            tags = [datetime.now().strftime('%Y%m%d%H%M%S')]

        print(f"开始构建镜像: {image_name}")

        # 克隆代码仓库（如果需要）
        if not os.path.exists(repo_path):
            print(f"克隆代码仓库: {repo_path}")
            subprocess.run(['git', 'clone', repo_url, repo_path], check=True)
        else:
            print(f"更新代码仓库: {repo_path}")
            subprocess.run(['git', 'pull'], cwd=repo_path, check=True)

        # 构建镜像
        for tag in tags:
            full_tag = f"{image_name}:{tag}"
            print(f"构建镜像标签: {full_tag}")

            image, build_logs = client.images.build(
                path=repo_path,
                tag=full_tag,
                rm=True,
                pull=True
            )

            # 输出构建日志
            for log in build_logs:
                if 'stream' in log:
                    print(log['stream'].strip())

            print(f"镜像 {full_tag} 构建成功")

        # 运行测试容器
        print("运行测试...")
        test_container = client.containers.run(
            image=f"{image_name}:{tags[0]}",
            command="pytest tests/",
            detach=True
        )

        # 等待测试完成
        test_result = test_container.wait()
        test_logs = test_container.logs().decode('utf-8')
        test_container.remove()

        if test_result['StatusCode'] != 0:
            print(f"测试失败: {test_result}")
            print(test_logs)
            return False

        print("测试成功")

        # 登录Docker仓库
        print("登录Docker仓库...")
        client.login(
            username=os.environ.get('DOCKER_USERNAME'),
            password=os.environ.get('DOCKER_PASSWORD'),
            registry=os.environ.get('DOCKER_REGISTRY', 'https://index.docker.io/v1/')
        )

        # 推送镜像
        for tag in tags:
            full_tag = f"{image_name}:{tag}"
            print(f"推送镜像: {full_tag}")

            push_logs = client.images.push(
                repository=image_name,
                tag=tag
            )

            print(push_logs)

        print(f"镜像 {image_name} 构建、测试和推送完成")
        return True

    except Exception as e:
        print(f"镜像构建流水线失败: {e}")
        return False

# 使用示例
build_and_push_image(
    repo_path='./my-app-repo',
    image_name='my-registry.com/my-app',
    tags=['v1.0.0', 'latest']
)

六、最佳实践与性能优化

6.1 高效使用Docker API

• 连接池管理：在高并发场景下，建议使用连接池管理Docker客户端连接，避免频繁创建新连接。
• 异步操作：对于I/O密集型操作，考虑使用docker库的异步API（需要安装aiohttp）：

import asyncio
import aiodocker

async def main():
    async with aiodocker.Docker() as docker:
        # 异步拉取镜像
        image = await docker.images.pull("python:3.10-slim")

        # 异步创建并启动容器
        container = await docker.containers.create_or_replace(
            config={
                "Image": "python:3.10-slim",
                "Cmd": ["python", "-c", "print('Hello, World!')"]
            },
            name="hello-world"
        )

        await container.start()
        logs = await container.log(stdout=True, stderr=True)
        print(logs)

        await container.delete(force=True)

asyncio.run(main())

6.2 镜像构建优化

• 使用多阶段构建：减少最终镜像大小，提高安全性。
• 缓存层优化：合理安排Dockerfile指令顺序，充分利用构建缓存。
• 避免安装不必要的包：只安装运行时必要的依赖。

6.3 容器资源管理

• 设置合理的资源限制：避免容器占用过多系统资源。

# 创建容器时设置资源限制
client.containers.create(
    image='my-app',
    name='resource-limited-app',
    mem_limit='512m',  # 内存限制
    memswap_limit='1g',  # 交换空间限制
    cpu_quota=50000,  # CPU配额（50%）
    detach=True
)

• 监控容器资源使用：定期收集容器资源使用数据，以便进行容量规划。

def monitor_container_resources(container_name):
    """监控容器资源使用情况"""
    container = client.containers.get(container_name)
    stats = container.stats(stream=False)

    # 解析CPU使用率
    cpu_percent = 0.0
    cpu_delta = float(stats['cpu_stats']['cpu_usage']['total_usage']) - float(stats['precpu_stats']['cpu_usage']['total_usage'])
    system_delta = float(stats['cpu_stats']['system_cpu_usage']) - float(stats['precpu_stats']['system_cpu_usage'])
    if system_delta > 0.0 and cpu_delta > 0.0:
        cpu_percent = (cpu_delta / system_delta) * len(stats['cpu_stats']['cpu_usage']['percpu_usage']) * 100.0

    # 解析内存使用率
    memory_usage = float(stats['memory_stats']['usage'])
    memory_limit = float(stats['memory_stats']['limit'])
    memory_percent = (memory_usage / memory_limit) * 100.0

    print(f"容器 {container_name} 资源使用:")
    print(f"CPU使用率: {cpu_percent:.2f}%")
    print(f"内存使用率: {memory_percent:.2f}% ({memory_usage/(1024*1024):.2f}MB / {memory_limit/(1024*1024):.2f}MB)")

    return {
        'cpu_percent': cpu_percent,
        'memory_percent': memory_percent,
        'memory_usage': memory_usage,
        'memory_limit': memory_limit
    }

# 监控容器资源
monitor_container_resources('python-app-container')

七、常见问题与解决方案

7.1 连接问题

• 问题：无法连接到Docker守护进程。
• 解决方案：

1. 确保Docker守护进程正在运行。
2. 检查DOCKER_HOST环境变量是否设置正确。
3. 对于远程连接，确保Docker守护进程配置为监听指定端口，并启用了适当的认证。

7.2 权限问题

• 问题：在Linux上运行时出现Permission denied错误。
• 解决方案：

1. 将当前用户添加到docker用户组：
   bash sudo usermod -aG docker $USER
2. 重新登录使更改生效。

7.3 镜像构建失败

• 问题：镜像构建过程中出现错误。
• 解决方案：

1. 检查Dockerfile语法是否正确。
2. 确保基础镜像存在且可访问。
3. 查看详细的构建日志，定位具体错误。
4. 使用docker build --no-cache强制重新构建所有层。

7.4 容器启动失败

• 问题：容器无法正常启动。
• 解决方案：

1. 使用container.wait()或查看容器日志获取详细错误信息。
2. 检查容器依赖的服务是否已启动。
3. 验证容器配置（端口映射、环境变量等）是否正确。

八、相关资源

• Pypi地址：https://pypi.org/project/docker/
• Github地址：https://github.com/docker/docker-py
• 官方文档地址：https://docker-py.readthedocs.io/en/stable/

关注我，每天分享一个实用的Python自动化工具。