如何用 Python 操作 Docker？
作者：肖恩頓
來源：游戲不存在
docker-py是Docker SDK for Python。docker-py主要利用了requests，使用http/socket協(xié)議連接本地的docker engine進(jìn)行操作。對 docker 感興趣，苦于工作中只用到 http 協(xié)議的同學(xué)，都建議閱讀一下本文。話不多說，一起了解docker-py的實現(xiàn)，本文分下面幾個部分:
docker-py項目結(jié)構(gòu) docker-py API示例 DockerClient的實現(xiàn) docker-version命令跟蹤 UnixHTTPAdapter的實現(xiàn) docker-ps命令跟蹤 docker-logs命令跟蹤 docker-exec 命令跟蹤 小結(jié) 小技巧
docker-py項目結(jié)構(gòu)
本次代碼閱讀，使用的版本是 4.2.0， 項目目錄結(jié)構(gòu)大概如下:
| 文件 | 描述 |
|---|---|
| client.py | docker客戶端的API |
| api | api相關(guān)目錄 |
| api/client.py | api的主要實現(xiàn) |
| api/container.py | container相關(guān)的api和client-mixin |
| api/daemon.py | daemon相關(guān)的api和client-mixin |
| models | 下為各種對象模型，主要是單體及集合 |
| models/resource.py | 模型基類 |
| models/containers.py | Container和ContainerCollection模型 |
| transport | 為客戶端和服務(wù)端的交互協(xié)議 |
| transport/unixconn.py | mac下主要使用了unix-sock實現(xiàn) |
還有一些目錄和類，因為不在這次介紹中，所以就沒有羅列。
docker-py API示例
docker-py API上手非常簡單:
import docker
client = docker.from_env()
result = client.version()
print(result)
# {'Platform': {'Name': 'Docker Engine - Community'},...}
client.containers.list()
# [<Container '45e6d2de7c54'>, <Container 'db18e4f20eaa'>, ...]
client.images.pull('nginx:1.10-alpine')
# <Image: 'nginx:1.10-alpine'>
client.images.list()
[<Image 'ubuntu'>, <Image 'nginx:1.10-alpine'>, ...]
上面示例展示了:
使用環(huán)境變量，創(chuàng)建client連接本地 docker-engine服務(wù)獲取版本號，等同 docker version獲取正在運(yùn)行的容器列表，等同 docker container list(別名是docker ps)拉取 nginx:1.10-alpin 鏡像，等同 docker image pull nginx:1.10-alpine(別名是docker pull nginx:1.10-alpine)獲取鏡像列表, 等同 docker image list
我們可以看到，docker-py的操作和docker的標(biāo)準(zhǔn)命令基本一致。
DockerClient的實現(xiàn)
DockerClient的構(gòu)造函數(shù)和工廠方法展示docker-client對象包裝了APIClient對象：
# client.py
class DockerClient(object):
def __init__(self, *args, **kwargs):
self.api = APIClient(*args, **kwargs)
@classmethod
def from_env(cls, **kwargs):
timeout = kwargs.pop('timeout', DEFAULT_TIMEOUT_SECONDS)
max_pool_size = kwargs.pop('max_pool_size', DEFAULT_MAX_POOL_SIZE)
version = kwargs.pop('version', None)
use_ssh_client = kwargs.pop('use_ssh_client', False)
return cls(
timeout=timeout,
max_pool_size=max_pool_size,
version=version,
use_ssh_client=use_ssh_client,
**kwargs_from_env(**kwargs)
)
DockerClient的API分2中，一種是屬性方法，比如常用的 containers，images，networks 和 volumes 等子命令，因為要將返回值包裝成對應(yīng)模型對象:
@property
def containers(self):
"""
An object for managing containers on the server. See the
:doc:`containers documentation <containers>` for full details.
"""
return ContainerCollection(client=self)
@property
def images(self):
return ImageCollection(client=self)
@property
def networks(self):
return NetworkCollection(client=self)
@property
def volumes(self):
return VolumeCollection(client=self)
...
另一種是不需要模型包裝，可以直接使用APIClient返回結(jié)果的 info, version 等方法:
# Top-level methods
def info(self, *args, **kwargs):
return self.api.info(*args, **kwargs)
info.__doc__ = APIClient.info.__doc__
def version(self, *args, **kwargs):
return self.api.version(*args, **kwargs)
version.__doc__ = APIClient.version.__doc__
...
DockerClient類工廠方法的全局引用:
from_env = DockerClient.from_env
docker-version命令跟蹤
我們先從簡單的 docker version 命令跟蹤查看APIClient如何工作的。APIClient的構(gòu)造函數(shù):
# api/client.py
import requests
class APIClient(
requests.Session,
BuildApiMixin,
ConfigApiMixin,
ContainerApiMixin,
DaemonApiMixin,
ExecApiMixin,
ImageApiMixin,
NetworkApiMixin,
PluginApiMixin,
SecretApiMixin,
ServiceApiMixin,
SwarmApiMixin,
VolumeApiMixin):
def __init__(self, base_url=None, version=None,
timeout=DEFAULT_TIMEOUT_SECONDS, tls=False,
user_agent=DEFAULT_USER_AGENT, num_pools=None,
credstore_env=None, use_ssh_client=False,
max_pool_size=DEFAULT_MAX_POOL_SIZE):
super(APIClient, self).__init__()
base_url = utils.parse_host(
base_url, IS_WINDOWS_PLATFORM, tls=bool(tls)
)
if base_url.startswith('http+unix://'):
self._custom_adapter = UnixHTTPAdapter(
base_url, timeout, pool_connections=num_pools,
max_pool_size=max_pool_size
)
self.mount('http+docker://', self._custom_adapter)
self._unmount('http://', 'https://')
# host part of URL should be unused, but is resolved by requests
# module in proxy_bypass_macosx_sysconf()
self.base_url = 'http+docker://localhost'
上面代碼可見:
APIClient繼承自 requests.Session APIClient使用Mixin方式組合了多個API，比如ContainerApiMixin提供container的api操作;NetWorkApiMixin提供network的api操作 使用mount方法加載不同協(xié)議的適配器adapter，unix系的docker是unix-socket;windows則是npipe
關(guān)于requests的使用，可以參看之前的博文 requests 源碼閱讀
默認(rèn)的服務(wù)URL實現(xiàn):
DEFAULT_UNIX_SOCKET = "http+unix:///var/run/docker.sock"
DEFAULT_NPIPE = 'npipe:////./pipe/docker_engine'
def parse_host(addr, is_win32=False, tls=False):
path = ''
port = None
host = None
# Sensible defaults
if not addr and is_win32:
return DEFAULT_NPIPE
if not addr or addr.strip() == 'unix://':
return DEFAULT_UNIX_SOCKET
version 請求在 DaemonApiMixin 中實現(xiàn):
class DaemonApiMixin(object):
def version(self, api_version=True):
url = self._url("/version", versioned_api=api_version)
return self._result(self._get(url), json=True)
底層的請求和響應(yīng)在主類APIClient中提供:
class APIClient
def _url(self, pathfmt, *args, **kwargs):
...
return '{0}{1}'.format(self.base_url, pathfmt.format(*args))
@update_headers
def _get(self, url, **kwargs):
return self.get(url, **self._set_request_timeout(kwargs))
def _result(self, response, json=False, binary=False):
assert not (json and binary)
self._raise_for_status(response)
if json:
return response.json()
if binary:
return response.content
return response.text
get和result，response都是requests提供。get發(fā)送請求，response.json將請求格式化成json后返回。
UnixHTTPAdapter的實現(xiàn)
/var/run/docker.sock是Docker守護(hù)程序偵聽的UNIX套接字,其連接使用UnixHTTPAdapter處理:
# transport/unixconn.py
import requests.adapters
RecentlyUsedContainer = urllib3._collections.RecentlyUsedContainer
class UnixHTTPAdapter(BaseHTTPAdapter):
def __init__(self, socket_url, timeout=60,
pool_connections=constants.DEFAULT_NUM_POOLS,
max_pool_size=constants.DEFAULT_MAX_POOL_SIZE):
socket_path = socket_url.replace('http+unix://', '')
if not socket_path.startswith('/'):
socket_path = '/' + socket_path
self.socket_path = socket_path
self.timeout = timeout
self.max_pool_size = max_pool_size
self.pools = RecentlyUsedContainer(
pool_connections, dispose_func=lambda p: p.close()
)
super(UnixHTTPAdapter, self).__init__()
def get_connection(self, url, proxies=None):
with self.pools.lock:
pool = self.pools.get(url)
if pool:
return pool
pool = UnixHTTPConnectionPool(
url, self.socket_path, self.timeout,
maxsize=self.max_pool_size
)
self.pools[url] = pool
return pool
UnixHTTPAdapter主要使用urllib3提供的鏈接池管理UnixHTTPConnection連接:
class UnixHTTPConnection(httplib.HTTPConnection, object):
def __init__(self, base_url, unix_socket, timeout=60):
super(UnixHTTPConnection, self).__init__(
'localhost', timeout=timeout
)
self.base_url = base_url
self.unix_socket = unix_socket
self.timeout = timeout
self.disable_buffering = False
def connect(self):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
sock.connect(self.unix_socket)
self.sock = sock
def putheader(self, header, *values):
super(UnixHTTPConnection, self).putheader(header, *values)
if header == 'Connection' and 'Upgrade' in values:
self.disable_buffering = True
def response_class(self, sock, *args, **kwargs):
if self.disable_buffering:
kwargs['disable_buffering'] = True
return UnixHTTPResponse(sock, *args, **kwargs)
class UnixHTTPConnectionPool(urllib3.connectionpool.HTTPConnectionPool):
def __init__(self, base_url, socket_path, timeout=60, maxsize=10):
super(UnixHTTPConnectionPool, self).__init__(
'localhost', timeout=timeout, maxsize=maxsize
)
self.base_url = base_url
self.socket_path = socket_path
self.timeout = timeout
def _new_conn(self):
return UnixHTTPConnection(
self.base_url, self.socket_path, self.timeout
)
connect展示了socket類型是 socket.AF_UNIX, 這一部分的實現(xiàn)都非?；A(chǔ) 。
關(guān)于socket，可以參看之前的博文 python http 源碼閱讀
docker-ps命令跟蹤
接著我們跟蹤稍微復(fù)雜點的命令 client.containers.list()， 也就是 docker ps。前面介紹了，container 會組裝結(jié)果為數(shù)據(jù)模型，下面是模型的父類:
class Model(object):
"""
A base class for representing a single object on the server.
"""
id_attribute = 'Id'
def __init__(self, attrs=None, client=None, collection=None):
self.client = client
# 集合
self.collection = collection
self.attrs = attrs
Model是單個模型抽象，Collection則是模型集合的抽象，使用集合的prepare_model構(gòu)建各種對象:
class Collection(object):
"""
A base class for representing all objects of a particular type on the
server.
"""
model = None
def __init__(self, client=None):
self.client = client
...
def prepare_model(self, attrs):
"""
Create a model from a set of attributes.
"""
if isinstance(attrs, Model):
attrs.client = self.client
# 雙向引用
attrs.collection = self
return attrs
elif isinstance(attrs, dict):
return self.model(attrs=attrs, client=self.client, collection=self)
else:
raise Exception("Can't create %s from %s" %
(self.model.__name__, attrs))
Container和ContainerCollection的實現(xiàn)
class Container(Model):
pass
class ContainerCollection(Collection):
model = Container
def get(self, container_id):
resp = self.client.api.inspect_container(container_id)
return self.prepare_model(resp)
def list(self, all=False, before=None, filters=None, limit=-1, since=None,
sparse=False, ignore_removed=False):
resp = self.client.api.containers(all=all, before=before,
filters=filters, limit=limit,
since=since)
containers = []
for r in resp:
containers.append(self.get(r['Id']))
return containers
其中l(wèi)ist函數(shù)主要有下面幾個步驟
使用api的containers接口得到resp，就是container-id列表 逐個循環(huán)使用api的inspect_container請求container的詳細(xì)信息 將結(jié)果封裝成Container對象 返回容器Container對象列表
api.containers和api.inspect_container在ContainerApiMixin中提供, 非常簡單清晰:
class ContainerApiMixin(object):
def containers(self, quiet=False, all=False, trunc=False, latest=False,
since=None, before=None, limit=-1, size=False,
filters=None):
params = {
'limit': 1 if latest else limit,
'all': 1 if all else 0,
'size': 1 if size else 0,
'trunc_cmd': 1 if trunc else 0,
'since': since,
'before': before
}
if filters:
params['filters'] = utils.convert_filters(filters)
u = self._url("/containers/json")
res = self._result(self._get(u, params=params), True)
if quiet:
return [{'Id': x['Id']} for x in res]
if trunc:
for x in res:
x['Id'] = x['Id'][:12]
return res
@utils.check_resource('container')
def inspect_container(self, container):
return self._result(
self._get(self._url("/containers/{0}/json", container)), True
)
docker-logs命令跟蹤
前面的命令都是request-response的模式，我們再看看不一樣的，基于流的docker-logs命令。我們先啟動一個容器:
docker run -d bfirsh/reticulate-splines
查看容器列表
# docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
61709b0ed4b8 bfirsh/reticulate-splines "/usr/local/bin/run.…" 22 seconds ago Up 21 seconds festive_pare
實時跟蹤容器運(yùn)行日志:
# docker logs -f 6170
Reticulating spline 1...
Reticulating spline 2...
....
可以看到reticulate-splines容器就是不停的打印行數(shù)數(shù)據(jù)。可以用下面的代碼實現(xiàn) docker logs 相同的功能:
logs = client.containers.get('61709b0ed4b8').logs(stream=True)
try:
while True:
line = next(logs).decode("utf-8")
print(line)
except StopIteration:
print(f'log stream ended for {container_name}')
代碼執(zhí)行結(jié)果和前面的類似:
# python sample.py
...
Reticulating spline 14...
Reticulating spline 15...
...
logs的實現(xiàn)中返回一個CancellableStream，而不是一個result，利用這個stream，就可以持續(xù)的讀取輸出:
# models/Container
def logs(self, **kwargs):
return self.client.api.logs(self.id, **kwargs)
# api/continer
def logs(self, container, stdout=True, stderr=True, stream=False,
timestamps=False, tail='all', since=None, follow=None,
until=None):
...
url = self._url("/containers/{0}/logs", container)
res = self._get(url, params=params, stream=stream)
output = self._get_result(container, stream, res)
if stream:
return CancellableStream(output, res)
else:
return output
比較特別的是下面對于stream的處理:
# api/client
def _multiplexed_response_stream_helper(self, response):
"""A generator of multiplexed data blocks coming from a response
stream."""
# Disable timeout on the underlying socket to prevent
# Read timed out(s) for long running processes
socket = self._get_raw_response_socket(response)
self._disable_socket_timeout(socket)
while True:
header = response.raw.read(STREAM_HEADER_SIZE_BYTES)
if not header:
break
_, length = struct.unpack('>BxxxL', header)
if not length:
continue
data = response.raw.read(length)
if not data:
break
yield data
def _disable_socket_timeout(self, socket):
sockets = [socket, getattr(socket, '_sock', None)]
for s in sockets:
if not hasattr(s, 'settimeout'):
continue
timeout = -1
if hasattr(s, 'gettimeout'):
timeout = s.gettimeout()
# Don't change the timeout if it is already disabled.
if timeout is None or timeout == 0.0:
continue
s.settimeout(None)
上面代碼展示了:
流的讀取方式是每次讀取STREAM_HEADER_SIZE_BYTES長度的數(shù)據(jù)作為協(xié)議頭 協(xié)議頭結(jié)構(gòu)體格式解壓后得到后面的數(shù)據(jù)包長度 繼續(xù)讀取指定長度的數(shù)據(jù)包 重復(fù)執(zhí)行上面的數(shù)據(jù)讀取過程 流式讀取的時候還需要關(guān)閉socket的超時機(jī)制，確保流一直保持，知道手動(ctl+c)關(guān)閉
而 attach 則是采用了websocket的實現(xiàn), 因為我們一般推薦使用exec命令，所以這里簡單了解即可:
def _attach_websocket(self, container, params=None):
url = self._url("/containers/{0}/attach/ws", container)
req = requests.Request("POST", url, params=self._attach_params(params))
full_url = req.prepare().url
full_url = full_url.replace("http://", "ws://", 1)
full_url = full_url.replace("https://", "wss://", 1)
return self._create_websocket_connection(full_url)
def _create_websocket_connection(self, url):
return websocket.create_connection(url)
docker-exec 命令跟蹤
docker-exec是我們的重頭戲，因為除了可以直接獲取docker是輸出外，還可以和docker進(jìn)行交互。先簡單回顧一下exec的使用:
# docker exec -it 2075 ping www.weibo.cn
PING www.weibo.cn (123.125.22.241): 56 data bytes
64 bytes from 123.125.22.241: seq=0 ttl=37 time=6.797 ms
64 bytes from 123.125.22.241: seq=1 ttl=37 time=39.279 ms
64 bytes from 123.125.22.241: seq=2 ttl=37 time=29.635 ms
64 bytes from 123.125.22.241: seq=3 ttl=37 time=27.737 ms
上面示例可以用下面代碼完全模擬:
result = client.containers.get("2075").exec_run("ping www.weibo.cn", tty=True, stream=True)
try:
while True:
line = next(result[1]).decode("utf-8")
print(line)
except StopIteration:
print(f'exec stream ended for {container_name}')
使用tty偽裝終端和容器進(jìn)行交互，就是我們最常用的方式了:
# docker exec -it 2075 sh
/ # ls -la
total 64
drwxr-xr-x 1 root root 4096 Mar 24 13:16 .
drwxr-xr-x 1 root root 4096 Mar 24 13:16 ..
-rwxr-xr-x 1 root root 0 Mar 24 13:16 .dockerenv
drwxr-xr-x 2 root root 4096 Mar 3 2017 bin
drwxr-xr-x 5 root root 340 Mar 24 13:16 dev
drwxr-xr-x 1 root root 4096 Mar 24 13:16 etc
drwxr-xr-x 2 root root 4096 Mar 3 2017 home
drwxr-xr-x 1 root root 4096 Mar 3 2017 lib
lrwxrwxrwx 1 root root 12 Mar 3 2017 linuxrc -> /bin/busybox
drwxr-xr-x 5 root root 4096 Mar 3 2017 media
drwxr-xr-x 2 root root 4096 Mar 3 2017 mnt
dr-xr-xr-x 156 root root 0 Mar 24 13:16 proc
drwx------ 1 root root 4096 Mar 25 08:17 root
drwxr-xr-x 2 root root 4096 Mar 3 2017 run
drwxr-xr-x 2 root root 4096 Mar 3 2017 sbin
drwxr-xr-x 2 root root 4096 Mar 3 2017 srv
dr-xr-xr-x 13 root root 0 Mar 24 13:16 sys
drwxrwxrwt 1 root root 4096 Mar 3 2017 tmp
drwxr-xr-x 1 root root 4096 Mar 3 2017 usr
drwxr-xr-x 1 root root 4096 Mar 3 2017 var
/ # exit
同樣這個過程也可以使用docker-py實現(xiàn):
_, socket = client.containers.get("2075").exec_run("sh", stdin=True, socket=True)
print(socket)
socket._sock.sendall(b"ls -la\n")
try:
unknown_byte=socket._sock.recv(docker.constants.STREAM_HEADER_SIZE_BYTES)
print(unknown_byte)
buffer_size = 4096 # 4 KiB
data = b''
while True:
part = socket._sock.recv(buffer_size)
data += part
if len(part) < buffer_size:
# either 0 or end of data
break
print(data.decode("utf8"))
except Exception:
pass
socket._sock.send(b"exit\n")
示例演示的過程是:
獲取一個已經(jīng)存在的容器2075 對容器執(zhí)行exec命令，注意需要開啟stdin和socket 向容器發(fā)送 ls -lah展示目錄列表讀區(qū)socket上的結(jié)果。(這里我們偷懶，沒有解析頭，直接硬取，這樣不夠健壯) 繼續(xù)發(fā)送 exit退出容器
程序的輸出和上面使用命令方式完全一致，就不在張貼了。進(jìn)入核心的exec_run函數(shù)的實現(xiàn):
# model/containers
def exec_run(self, cmd, stdout=True, stderr=True, stdin=False, tty=False,
privileged=False, user='', detach=False, stream=False,
socket=False, environment=None, workdir=None, demux=False):
resp = self.client.api.exec_create(
self.id, cmd, stdout=stdout, stderr=stderr, stdin=stdin, tty=tty,
privileged=privileged, user=user, environment=environment,
workdir=workdir,
)
exec_output = self.client.api.exec_start(
resp['Id'], detach=detach, tty=tty, stream=stream, socket=socket,
demux=demux
)
if socket or stream:
return ExecResult(None, exec_output)
主要使用API的exec_create和exec_start兩個函數(shù), 先看第一個exec_create函數(shù):
# api/exec_api
def exec_create(self, container, cmd, stdout=True, stderr=True,
stdin=False, tty=False, privileged=False, user='',
environment=None, workdir=None, detach_keys=None):
if isinstance(cmd, six.string_types):
cmd = utils.split_command(cmd)
if isinstance(environment, dict):
environment = utils.utils.format_environment(environment)
data = {
'Container': container,
'User': user,
'Privileged': privileged,
'Tty': tty,
'AttachStdin': stdin,
'AttachStdout': stdout,
'AttachStderr': stderr,
'Cmd': cmd,
'Env': environment,
}
if detach_keys:
data['detachKeys'] = detach_keys
elif 'detachKeys' in self._general_configs:
data['detachKeys'] = self._general_configs['detachKeys']
url = self._url("/containers/{0}/exec", container)
res = self._post_json(url, data=data)
return self._result(res, True)
exec_create相對還是比較簡單，就是post-json數(shù)據(jù)到 /containers/{0}/exec 接口。然后是exec_start函數(shù):
def exec_start(self, exec_id, detach=False, tty=False, stream=False,
socket=False, demux=False):
# we want opened socket if socket == True
data = {
'Tty': tty,
'Detach': detach
}
headers = {} if detach else {
'Connection': 'Upgrade',
'Upgrade': 'tcp'
}
res = self._post_json(
self._url("/exec/{0}/start", exec_id),
headers=headers,
data=data,
stream=True
)
if detach:
return self._result(res)
if socket:
return self._get_raw_response_socket(res)
return self._read_from_socket(res, stream, tty=tty, demux=demux)
exec_start是post-json到 /exec/{0}/start 接口，注意這個接口看起來不是到容器，而是到exec。然后如果socket參數(shù)是true則返回socket，可以進(jìn)行寫入；否則僅僅讀取數(shù)據(jù)。
使用curl訪問docker-api
docker-engine的REST-api也可以直接使用 curl 訪問:
$ curl --unix-socket /var/run/docker.sock -H "Content-Type: application/json" \
-d '{"Image": "alpine", "Cmd": ["echo", "hello world"]}' \
-X POST http://localhost/v1.41/containers/create
{"Id":"1c6594faf5","Warnings":null}
$ curl --unix-socket /var/run/docker.sock -X POST http://localhost/v1.41/containers/1c6594faf5/start
$ curl --unix-socket /var/run/docker.sock -X POST http://localhost/v1.41/containers/1c6594faf5/wait
{"StatusCode":0}
$ curl --unix-socket /var/run/docker.sock "http://localhost/v1.41/containers/1c6594faf5/logs?stdout=1"
hello world
可以通過修改/etc/docker/daemon.json更改為http服務(wù)方式的api
{
"debug": true,
"hosts": ["tcp://192.168.59.3:2376"]
}
然后 curl 命令可以直接訪問docker的api
curl http://127.0.0.1:2375/info
curl http://127.0.0.1:2375/version
curl http://127.0.0.1:2375/images/json
curl http://127.0.0.1:2375/images/alpine/json
curl http://127.0.0.1:2375/containers/json
curl http://127.0.0.1:2375/containers/25c5805a06b6/json
小結(jié)
利用docker-py可以完全操作docker，這得益docker提供的REST-api操作。同時也發(fā)現(xiàn)requests的設(shè)計很強(qiáng)大，不僅僅可以用來做http請求，還可以用來做socket請求。學(xué)習(xí)docker-py后，相信大家對docker的理解一定有那么一點點加深，也希望下面這張圖可以幫助你記憶:
小技巧
使用 check_resource 裝飾器，對函數(shù)的參數(shù)進(jìn)行預(yù)先處理:
def check_resource(resource_name):
def decorator(f):
@functools.wraps(f)
def wrapped(self, resource_id=None, *args, **kwargs):
if resource_id is None and kwargs.get(resource_name):
resource_id = kwargs.pop(resource_name)
if isinstance(resource_id, dict):
resource_id = resource_id.get('Id', resource_id.get('ID'))
if not resource_id:
raise errors.NullResource(
'Resource ID was not provided'
)
return f(self, resource_id, *args, **kwargs)
return wrapped
return decorator
代碼版本比較工具:
from distutils.version import StrictVersion
def compare_version(v1, v2):
"""Compare docker versions
>>> v1 = '1.9'
>>> v2 = '1.10'
>>> compare_version(v1, v2)
1
>>> compare_version(v2, v1)
-1
>>> compare_version(v2, v2)
0
"""
s1 = StrictVersion(v1)
s2 = StrictVersion(v2)
if s1 == s2:
return 0
elif s1 > s2:
return -1
else:
return 1
def version_lt(v1, v2):
return compare_version(v1, v2) > 0
def version_gte(v1, v2):
return not version_lt(v1, v2)
參考鏈接
https://docs.docker.com/engine/api/sdk/examples/ https://docker-py.readthedocs.io/en/stable/
近期熱門文章推薦：
評論
圖片
表情