mirror of https://github.com/docker/docker-py.git
108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
import json
|
|
|
|
from ..errors import InvalidVersion
|
|
from ..utils import check_resource, minimum_version
|
|
from ..utils import version_lt
|
|
|
|
|
|
class NetworkApiMixin(object):
|
|
@minimum_version('1.21')
|
|
def networks(self, names=None, ids=None):
|
|
filters = {}
|
|
if names:
|
|
filters['name'] = names
|
|
if ids:
|
|
filters['id'] = ids
|
|
|
|
params = {'filters': json.dumps(filters)}
|
|
|
|
url = self._url("/networks")
|
|
res = self._get(url, params=params)
|
|
return self._result(res, json=True)
|
|
|
|
@minimum_version('1.21')
|
|
def create_network(self, name, driver=None, options=None, ipam=None,
|
|
check_duplicate=None, internal=False, labels=None,
|
|
enable_ipv6=False):
|
|
if options is not None and not isinstance(options, dict):
|
|
raise TypeError('options must be a dictionary')
|
|
|
|
data = {
|
|
'Name': name,
|
|
'Driver': driver,
|
|
'Options': options,
|
|
'IPAM': ipam,
|
|
'CheckDuplicate': check_duplicate
|
|
}
|
|
|
|
if labels is not None:
|
|
if version_lt(self._version, '1.23'):
|
|
raise InvalidVersion(
|
|
'network labels were introduced in API 1.23'
|
|
)
|
|
if not isinstance(labels, dict):
|
|
raise TypeError('labels must be a dictionary')
|
|
data["Labels"] = labels
|
|
|
|
if enable_ipv6:
|
|
if version_lt(self._version, '1.23'):
|
|
raise InvalidVersion(
|
|
'enable_ipv6 was introduced in API 1.23'
|
|
)
|
|
data['EnableIPv6'] = True
|
|
|
|
if internal:
|
|
if version_lt(self._version, '1.22'):
|
|
raise InvalidVersion('Internal networks are not '
|
|
'supported in API version < 1.22')
|
|
data['Internal'] = True
|
|
|
|
url = self._url("/networks/create")
|
|
res = self._post_json(url, data=data)
|
|
return self._result(res, json=True)
|
|
|
|
@minimum_version('1.21')
|
|
def remove_network(self, net_id):
|
|
url = self._url("/networks/{0}", net_id)
|
|
res = self._delete(url)
|
|
self._raise_for_status(res)
|
|
|
|
@minimum_version('1.21')
|
|
def inspect_network(self, net_id):
|
|
url = self._url("/networks/{0}", net_id)
|
|
res = self._get(url)
|
|
return self._result(res, json=True)
|
|
|
|
@check_resource
|
|
@minimum_version('1.21')
|
|
def connect_container_to_network(self, container, net_id,
|
|
ipv4_address=None, ipv6_address=None,
|
|
aliases=None, links=None,
|
|
link_local_ips=None):
|
|
data = {
|
|
"Container": container,
|
|
"EndpointConfig": self.create_endpoint_config(
|
|
aliases=aliases, links=links, ipv4_address=ipv4_address,
|
|
ipv6_address=ipv6_address, link_local_ips=link_local_ips
|
|
),
|
|
}
|
|
|
|
url = self._url("/networks/{0}/connect", net_id)
|
|
res = self._post_json(url, data=data)
|
|
self._raise_for_status(res)
|
|
|
|
@check_resource
|
|
@minimum_version('1.21')
|
|
def disconnect_container_from_network(self, container, net_id,
|
|
force=False):
|
|
data = {"Container": container}
|
|
if force:
|
|
if version_lt(self._version, '1.22'):
|
|
raise InvalidVersion(
|
|
'Forced disconnect was introduced in API 1.22'
|
|
)
|
|
data['Force'] = force
|
|
url = self._url("/networks/{0}/disconnect", net_id)
|
|
res = self._post_json(url, data=data)
|
|
self._raise_for_status(res)
|