Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def engine_max_version():
if 'DOCKER_VERSION' not in os.environ:
return V3_5
version = os.environ['DOCKER_VERSION'].partition('-')[0]
if version_lt(version, '1.10'):
return V1
if version_lt(version, '1.12'):
return V2_0
if version_lt(version, '1.13'):
return V2_1
if version_lt(version, '17.06'):
return V3_2
return V3_5
def engine_max_version():
if 'DOCKER_VERSION' not in os.environ:
return V3_3
version = os.environ['DOCKER_VERSION'].partition('-')[0]
if version_lt(version, '1.10'):
return V1
if version_lt(version, '1.12'):
return V2_0
if version_lt(version, '1.13'):
return V2_1
if version_lt(version, '17.06'):
return V3_2
return V3_3
container_options['volumes'] = dict((v.internal, {}) for v in container_volumes or {})
if version_gte(self.client.api_version, '1.30'):
override_options['mounts'] = [build_mount(v) for v in container_mounts] or None
else:
# Workaround for 3.2 format
override_options['tmpfs'] = self.options.get('tmpfs') or []
for m in container_mounts:
if m.is_tmpfs:
override_options['tmpfs'].append(m.target)
else:
binds.append(m.legacy_repr())
container_options['volumes'][m.target] = {}
secret_volumes = self.get_secret_volumes()
if secret_volumes:
if version_lt(self.client.api_version, '1.30'):
binds.extend(v.legacy_repr() for v in secret_volumes)
container_options['volumes'].update(
(v.target, {}) for v in secret_volumes
)
else:
override_options['mounts'] = override_options.get('mounts') or []
override_options['mounts'].extend([build_mount(v) for v in secret_volumes])
# Remove possible duplicates (see e.g. https://github.com/docker/compose/issues/5885).
# unique_everseen preserves order. (see https://github.com/docker/compose/issues/6091).
override_options['binds'] = list(unique_everseen(binds))
return container_options, override_options
Returns (dict): ID of the newly created secret
"""
if not isinstance(data, bytes):
data = data.encode('utf-8')
data = base64.b64encode(data)
if six.PY3:
data = data.decode('ascii')
body = {
'Data': data,
'Name': name,
'Labels': labels
}
if driver is not None:
if utils.version_lt(self._version, '1.31'):
raise errors.InvalidVersion(
'Secret driver is only available for API version > 1.31'
)
body['Driver'] = driver
url = self._url('/secrets/create')
return self._result(
self._post_json(url, data=body), True
)
def __init__(self, version, aliases=None, links=None, ipv4_address=None,
ipv6_address=None, link_local_ips=None):
if version_lt(version, '1.22'):
raise errors.InvalidVersion(
'Endpoint config is not supported for API version < 1.22'
)
if aliases:
self["Aliases"] = aliases
if links:
self["Links"] = normalize_links(links)
ipam_config = {}
if ipv4_address:
ipam_config['IPv4Address'] = ipv4_address
if ipv6_address:
ipam_config['IPv6Address'] = ipv6_address
def events(self, service_names=None):
if version_lt(self.client.api_version, '1.22'):
# New, better event API was introduced in 1.22.
return self._legacy_event_processor(service_names)
def build_container_event(event):
container_attrs = event['Actor']['Attributes']
time = datetime.datetime.fromtimestamp(event['time'])
time = time.replace(
microsecond=microseconds_from_time_nano(event['timeNano'])
)
container = None
try:
container = Container.from_id(self.client, event['id'])
except APIError:
# Container may have been removed (e.g. if this is a destroy event)
pass
container_options = dict(
(k, self.options[k])
for k in DOCKER_CONFIG_KEYS if k in self.options)
override_volumes = override_options.pop('volumes', [])
container_options.update(override_options)
if not container_options.get('name'):
container_options['name'] = self.get_container_name(self.name, number, slug)
container_options.setdefault('detach', True)
# If a qualified hostname was given, split it into an
# unqualified hostname and a domainname unless domainname
# was also given explicitly. This matches behavior
# until Docker Engine 1.11.0 - Docker API 1.23.
if (version_lt(self.client.api_version, '1.23') and
'hostname' in container_options and
'domainname' not in container_options and
'.' in container_options['hostname']):
parts = container_options['hostname'].partition('.')
container_options['hostname'] = parts[0]
container_options['domainname'] = parts[2]
if (version_gte(self.client.api_version, '1.25') and
'stop_grace_period' in self.options):
container_options['stop_timeout'] = self.stop_timeout(None)
if 'ports' in container_options or 'expose' in self.options:
container_options['ports'] = build_container_ports(
formatted_ports(container_options.get('ports', [])),
self.options)
service (str): Service name or ID.
insert_defaults (boolean): If true, default values will be merged
into the service inspect output.
Returns:
(dict): A dictionary of the server-side representation of the
service, including all relevant properties.
Raises:
:py:class:`docker.errors.APIError`
If the server returns an error.
"""
url = self._url('/services/{0}', service)
params = {}
if insert_defaults is not None:
if utils.version_lt(self._version, '1.29'):
raise errors.InvalidVersion(
'insert_defaults is not supported in API version < 1.29'
)
params['insertDefaults'] = insert_defaults
return self._result(self._get(url, params=params), True)
def disconnect_container_from_network(self, container, net_id,
force=False):
"""
Disconnect a container from a network.
Args:
container (str): container ID or name to be disconnected from the
network
net_id (str): network ID
force (bool): Force the container to disconnect from a network.
Default: ``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)