上一篇 下一篇 分享链接 返回 返回顶部

Docker API 实战:从接口调用到 Python 管理容器源码示例

发布人:慈云数据-客服中心 发布时间:8小时前 阅读量:1

Docker API接口调用教程|附源码

Docker 不仅可以通过命令行管理容器,还提供了功能完整的 Remote API(远程接口),可以让我们用程序直接控制 Docker Engine。
无论你想做容器自动化运维、Web 管理平台、CI/CD 集成,还是开发自己的 Docker 管理工具,掌握 Docker API 都非常实用。

本文将从 Docker API 的基础概念、开启方式、常见接口、调用示例、Python 源码实战 等方面,带你系统理解如何调用 Docker API。


一、什么是 Docker API

Docker API 是 Docker Daemon 提供的一组 HTTP 接口。
你可以通过它完成几乎所有 Docker 命令行能做的事情,例如:

  • 查看镜像列表
  • 创建和启动容器
  • 停止和删除容器
  • 拉取镜像
  • 查看日志
  • 查询容器状态
  • 执行容器命令

本质上,docker 命令行工具也是通过 Docker API 与守护进程通信的。


二、Docker API 的访问方式

Docker API 通常有两种访问方式:

1)本地 Unix Socket

在 Linux/macOS 上,Docker 默认通过:

/var/run/docker.sock

进行通信。

这种方式最常见,性能也最好。
如果你在本机运行程序,只要能访问这个 socket,就能调用 Docker API。

2)TCP 端口

Docker 也可以监听 TCP 端口,例如:

tcp://127.0.0.1:2375

或者启用 TLS 后使用:

tcp://your-docker-host:2376

注意:直接开放 2375 端口是有安全风险的,生产环境务必使用 TLS 或其他访问控制手段。


三、开启 Docker API 访问

方式一:查看当前 Docker 是否支持 socket

在 Linux 上执行:

ls -l /var/run/docker.sock

如果存在,说明 Docker 已经通过 socket 提供接口访问。


方式二:配置 Docker 监听 TCP

如果你需要远程调用 Docker API,可以修改 Docker 配置。

1. 修改 systemd 配置

编辑 Docker 服务配置:

sudo vim /lib/systemd/system/docker.service

找到类似这一行:

ExecStart=/usr/bin/dockerd -H fd://

改成:

ExecStart=/usr/bin/dockerd -H fd:// -H tcp://0.0.0.0:2375

然后重载并重启:

sudo systemctl daemon-reload
sudo systemctl restart docker

2. 验证接口

执行:

curl http://127.0.0.1:2375/version

如果返回 JSON,说明 API 已可用。


四、Docker API 常见版本

Docker API 有版本号,例如:

  • v1.41
  • v1.42
  • v1.43

建议在请求时带上版本号,例如:

http://localhost:2375/v1.41/containers/json

这样可以避免不同 Docker 版本之间接口不兼容。


五、常用 Docker API 接口

下面列出一些最常见的接口:

1)查看 Docker 版本

GET /version

示例:

curl http://127.0.0.1:2375/version

2)列出容器

GET /containers/json

示例:

curl http://127.0.0.1:2375/v1.41/containers/json

参数说明:

  • all=1:显示所有容器,包括已停止的
  • size=1:显示容器大小

示例:

curl "http://127.0.0.1:2375/v1.41/containers/json?all=1"

3)查看容器详情

GET /containers/{id}/json

示例:

curl http://127.0.0.1:2375/v1.41/containers//json

4)启动容器

POST /containers/create
POST /containers/{id}/start

一般分两步:

  1. 创建容器
  2. 启动容器

5)停止容器

POST /containers/{id}/stop

6)删除容器

DELETE /containers/{id}

7)拉取镜像

POST /images/create?fromImage=nginx&tag=latest

8)查看日志

GET /containers/{id}/logs

六、用 curl 调用 Docker API

先来看最直接的方式:curl

1. 查看版本

curl http://127.0.0.1:2375/version

2. 查看容器列表

curl http://127.0.0.1:2375/v1.41/containers/json?all=1

3. 拉取 nginx 镜像

curl -X POST "http://127.0.0.1:2375/v1.41/images/create?fromImage=nginx&tag=latest"

4. 创建容器

curl -X POST http://127.0.0.1:2375/v1.41/containers/create \
  -H "Content-Type: application/json" \
  -d '{
    "Image": "nginx:latest",
    "Name": "my-nginx",
    "HostConfig": {
      "PortBindings": {
        "80/tcp": [
          {
            "HostPort": "8080"
          }
        ]
      }
    }
  }'

返回结果中会包含容器 ID。


5. 启动容器

假设容器 ID 为 abc123...

curl -X POST http://127.0.0.1:2375/v1.41/containers/abc123/start

6. 停止容器

curl -X POST http://127.0.0.1:2375/v1.41/containers/abc123/stop

七、Python 调用 Docker API 实战

下面给出一个完整的 Python 示例,使用 requests 调用 Docker API。

1)安装依赖

pip install requests

2)项目结构

docker-api-demo/
├── docker_api_demo.py
└── requirements.txt

3)requirements.txt

requests==2.32.3

4)源码:docker_api_demo.py

import requests
import json

BASE_URL = "http://127.0.0.1:2375/v1.41"


class DockerAPIClient:
    def __init__(self, base_url=BASE_URL):
        self.base_url = base_url.rstrip("/")

    def _get(self, path, params=None):
        url = f"{self.base_url}{path}"
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        return resp.json() if resp.text else None

    def _post(self, path, params=None, data=None, json_data=None):
        url = f"{self.base_url}{path}"
        resp = requests.post(url, params=params, data=data, json=json_data, timeout=10)
        resp.raise_for_status()
        return resp.json() if resp.text else None

    def _delete(self, path, params=None):
        url = f"{self.base_url}{path}"
        resp = requests.delete(url, params=params, timeout=10)
        resp.raise_for_status()
        return resp.json() if resp.text else None

    def version(self):
        # Docker 版本接口通常不需要 API 版本前缀
        url = self.base_url.split("/v1.41")[0] + "/version"
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
        return resp.json()

    def list_containers(self, all=True):
        params = {"all": 1} if all else None
        return self._get("/containers/json", params=params)

    def pull_image(self, image="nginx", tag="latest"):
        params = {"fromImage": image, "tag": tag}
        url = f"{self.base_url}/images/create"
        resp = requests.post(url, params=params, stream=True, timeout=60)

        # 拉取镜像时 Docker 会分块返回日志流
        resp.raise_for_status()
        for line in resp.iter_lines():
            if line:
                print(line.decode("utf-8"))

    def create_container(self, image="nginx:latest", name="my-nginx", host_port="8080"):
        payload = {
            "Image": image,
            "HostConfig": {
                "PortBindings": {
                    "80/tcp": [
                        {
                            "HostPort": host_port
                        }
                    ]
                }
            }
        }

        params = {"name": name}
        return self._post("/containers/create", params=params, json_data=payload)

    def start_container(self, container_id):
        url = f"{self.base_url}/containers/{container_id}/start"
        resp = requests.post(url, timeout=10)
        resp.raise_for_status()
        return True

    def stop_container(self, container_id):
        url = f"{self.base_url}/containers/{container_id}/stop"
        resp = requests.post(url, timeout=10)
        resp.raise_for_status()
        return True

    def remove_container(self, container_id, force=False):
        params = {"force": "1"} if force else None
        return self._delete(f"/containers/{container_id}", params=params)

    def inspect_container(self, container_id):
        return self._get(f"/containers/{container_id}/json")


if __name__ == "__main__":
    client = DockerAPIClient()

    print("=== Docker Version ===")
    print(json.dumps(client.version(), indent=2, ensure_ascii=False))

    print("\n=== Container List ===")
    containers = client.list_containers(all=True)
    print(json.dumps(containers, indent=2, ensure_ascii=False))

    print("\n=== Pull nginx Image ===")
    client.pull_image("nginx", "latest")

    print("\n=== Create Container ===")
    result = client.create_container(image="nginx:latest", name="my-nginx", host_port="8080")
    print(json.dumps(result, indent=2, ensure_ascii=False))

    container_id = result.get("Id")
    if container_id:
        print("\n=== Start Container ===")
        client.start_container(container_id)
        print("Container started")

        print("\n=== Inspect Container ===")
        detail = client.inspect_container(container_id)
        print(json.dumps(detail, indent=2, ensure_ascii=False))

        print("\n=== Stop Container ===")
        client.stop_container(container_id)
        print("Container stopped")

        print("\n=== Remove Container ===")
        client.remove_container(container_id, force=True)
        print("Container removed")

八、代码说明

这份源码实现了几个基础功能:

1. version()

获取 Docker Engine 版本信息。

2. list_containers()

列出所有容器,支持查看运行中和已停止的容器。

3. pull_image()

拉取镜像。
这个接口会持续返回流式数据,因此使用了 stream=True

4. create_container()

创建一个 Nginx 容器,并映射端口:

  • 容器端口:80
  • 宿主机端口:8080

5. start_container() / stop_container()

控制容器启停。

6. inspect_container()

查看容器详细信息。

7. remove_container()

删除容器。


九、运行效果示例

执行脚本后,可能看到如下输出:

=== Docker Version ===
{
  "Version": "24.x.x",
  "ApiVersion": "1.41",
  ...
}

=== Container List ===
[
  {
    "Id": "xxxx",
    "Image": "nginx:latest",
    "Names": ["/my-nginx"],
    ...
  }
]

=== Pull nginx Image ===
{"status":"Pulling from library/nginx"}
{"status":"Downloading"}
{"status":"Extracting"}

=== Create Container ===
{
  "Id": "abc123...",
  "Warnings": null
}

=== Start Container ===
Container started

十、Docker API 调用中的注意事项

1)不要随意暴露 2375 端口

如果 Docker API 暴露在公网,攻击者可以直接控制宿主机上的容器,风险极高。

2)建议使用 TLS

生产环境推荐启用:

  • 客户端证书
  • 服务端证书
  • 双向认证

3)给程序最小权限

如果你的程序只是查询容器状态,不要赋予过高权限。

4)注意 API 版本兼容

不同 Docker 版本的 API 可能存在差异,建议在代码中固定 API 版本。

5)处理流式返回

像拉镜像、查看日志、事件监听等接口,返回内容可能是流式的,需要特殊处理。


十一、适合哪些场景

Docker API 特别适合以下场景:

  • 自动部署平台
  • 容器监控系统
  • CI/CD 构建流程
  • 多租户容器管理后台
  • 资源编排工具
  • 一键创建测试环境

比如你可以开发一个 Web 页面,点击按钮就创建一个新容器;
也可以编写运维脚本,定时扫描无用容器并自动清理。


十二、进阶方向

如果你已经掌握了基础调用,可以继续学习以下内容:

1. 事件监听

通过 /events 接口监听 Docker 事件,实现容器状态实时监控。

2. 容器日志实时读取

通过 /containers/{id}/logs?follow=1 获取实时日志。

3. 执行容器命令

通过 exec/createexec/start 在容器内部执行命令。

4. Swarm 与服务管理

如果你使用 Docker Swarm,还可以调用服务编排相关 API。

5. 封装成管理平台

将 Docker API 封装为 REST 服务,提供给前端或其他系统调用。


十三、总结

Docker API 是 Docker 自动化管理的核心接口。
相比手工执行 docker rundocker stop 等命令,API 调用更适合程序化、批量化、平台化的场景。

本文介绍了:

  • Docker API 的基本概念
  • socket 与 TCP 两种访问方式
  • 常用接口
  • curl 调用示例
  • Python 完整源码示例
  • 安全注意事项

如果你想构建自己的容器管理系统,这套接口是必学内容。


如果你愿意,我还可以继续为你补充以下内容之一:

  1. Docker API 调用 Go 语言版源码
  2. Docker API 事件监听教程
  3. Docker API + Flask 实现一个容器管理后台
  4. Docker API TLS 安全认证配置教程

如果需要,我可以直接接着写下一篇。

目录结构
全文