Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def ports(self):
"""
:return: dict
{
# container -> host
"1234": "2345"
}
"""
if self._ports is None:
self._ports = {}
if self.net_settings["Ports"]:
for key, value in self.net_settings["Ports"].items():
cleaned_port = key.split("/")[0]
self._ports[cleaned_port] = graceful_chain_get(value, 0, "HostPort")
# in case of --net=host, there's nothing in network settings, let's get it from "Config"
exposed_ports_section = graceful_chain_get(self.inspect_data, "Config", "ExposedPorts")
if exposed_ports_section:
for key, value in exposed_ports_section.items():
cleaned_port = key.split("/")[0]
self._ports[cleaned_port] = None # extremely docker specific
return self._ports
def __init__(self, inspect_data):
self.inspect_data = inspect_data
self.net_settings = graceful_chain_get(self.inspect_data, "NetworkSettings")
self._ports = None
self._ips = None
def df(self, cached=True):
if cached is False or self._df is None:
logger.debug("getting disk-usage")
# TODO: wrap in try/execpt
self._df = self.client.df()
# TODO: attach these to real containers and images
# # since DOCKER API-1.25 (v.1.13.0)
# df = self.client.df()
# if 'Containers' in df:
# df_containers = df['Containers']
containers_data = graceful_chain_get(self._df, "Containers")
for c_data in containers_data:
c = graceful_chain_get(self._containers, graceful_chain_get(c_data, "Id"))
c.size_root_fs = graceful_chain_get(c_data, "SizeRootFs")
c.size_rw_fs = graceful_chain_get(c_data, "SizeRw")
images_data = graceful_chain_get(self._df, "Images")
for i_data in images_data:
i = graceful_chain_get(self._images, graceful_chain_get(i_data, "Id"))
i._total_size = graceful_chain_get(i_data, "Size")
i._shared_size = graceful_chain_get(i_data, "SharedSize")
i._virtual_size = graceful_chain_get(i_data, "VirtualSize")
return self._df
def df(self, cached=True):
if cached is False or self._df is None:
logger.debug("getting disk-usage")
# TODO: wrap in try/execpt
self._df = self.client.df()
# TODO: attach these to real containers and images
# # since DOCKER API-1.25 (v.1.13.0)
# df = self.client.df()
# if 'Containers' in df:
# df_containers = df['Containers']
containers_data = graceful_chain_get(self._df, "Containers")
for c_data in containers_data:
c = graceful_chain_get(self._containers, graceful_chain_get(c_data, "Id"))
c.size_root_fs = graceful_chain_get(c_data, "SizeRootFs")
c.size_rw_fs = graceful_chain_get(c_data, "SizeRw")
images_data = graceful_chain_get(self._df, "Images")
for i_data in images_data:
i = graceful_chain_get(self._images, graceful_chain_get(i_data, "Id"))
i._total_size = graceful_chain_get(i_data, "Size")
i._shared_size = graceful_chain_get(i_data, "SharedSize")
i._virtual_size = graceful_chain_get(i_data, "VirtualSize")
return self._df
:param network_name: str
:param network_data: dict
:return: dict:
{
"ip_address4": "12.34.56.78"
"ip_address6": "ff:fa:..."
}
"""
a4 = None
if network_name == "host":
a4 = "127.0.0.1"
n = {}
a4 = graceful_chain_get(network_data, "IPAddress") or a4
if a4:
n["ip_address4"] = a4
a6 = graceful_chain_get(network_data, "GlobalIPv6Address")
if a6:
n["ip_address4"] = a6
return n
def df(self, cached=True):
if cached is False or self._df is None:
logger.debug("getting disk-usage")
# TODO: wrap in try/execpt
self._df = self.client.df()
# TODO: attach these to real containers and images
# # since DOCKER API-1.25 (v.1.13.0)
# df = self.client.df()
# if 'Containers' in df:
# df_containers = df['Containers']
containers_data = graceful_chain_get(self._df, "Containers")
for c_data in containers_data:
c = graceful_chain_get(self._containers, graceful_chain_get(c_data, "Id"))
c.size_root_fs = graceful_chain_get(c_data, "SizeRootFs")
c.size_rw_fs = graceful_chain_get(c_data, "SizeRw")
images_data = graceful_chain_get(self._df, "Images")
for i_data in images_data:
i = graceful_chain_get(self._images, graceful_chain_get(i_data, "Id"))
i._total_size = graceful_chain_get(i_data, "Size")
i._shared_size = graceful_chain_get(i_data, "SharedSize")
i._virtual_size = graceful_chain_get(i_data, "VirtualSize")
return self._df
def df(self, cached=True):
if cached is False or self._df is None:
logger.debug("getting disk-usage")
# TODO: wrap in try/execpt
self._df = self.client.df()
# TODO: attach these to real containers and images
# # since DOCKER API-1.25 (v.1.13.0)
# df = self.client.df()
# if 'Containers' in df:
# df_containers = df['Containers']
containers_data = graceful_chain_get(self._df, "Containers")
for c_data in containers_data:
c = graceful_chain_get(self._containers, graceful_chain_get(c_data, "Id"))
c.size_root_fs = graceful_chain_get(c_data, "SizeRootFs")
c.size_rw_fs = graceful_chain_get(c_data, "SizeRw")
images_data = graceful_chain_get(self._df, "Images")
for i_data in images_data:
i = graceful_chain_get(self._images, graceful_chain_get(i_data, "Id"))
i._total_size = graceful_chain_get(i_data, "Size")
i._shared_size = graceful_chain_get(i_data, "SharedSize")
i._virtual_size = graceful_chain_get(i_data, "VirtualSize")
return self._df
def metadata_get(self, path, cached=True):
"""
get metadata from inspect, specified by path
:param path: list of str
:param cached: bool, use cached version of inspect if available
"""
try:
value = graceful_chain_get(self.inspect(cached=cached).response, *path)
except docker.errors.NotFound:
logger.warning("object %s is not available anymore", self)
raise NotAvailableAnymore()
return value
logger.debug("getting disk-usage")
# TODO: wrap in try/execpt
self._df = self.client.df()
# TODO: attach these to real containers and images
# # since DOCKER API-1.25 (v.1.13.0)
# df = self.client.df()
# if 'Containers' in df:
# df_containers = df['Containers']
containers_data = graceful_chain_get(self._df, "Containers")
for c_data in containers_data:
c = graceful_chain_get(self._containers, graceful_chain_get(c_data, "Id"))
c.size_root_fs = graceful_chain_get(c_data, "SizeRootFs")
c.size_rw_fs = graceful_chain_get(c_data, "SizeRw")
images_data = graceful_chain_get(self._df, "Images")
for i_data in images_data:
i = graceful_chain_get(self._images, graceful_chain_get(i_data, "Id"))
i._total_size = graceful_chain_get(i_data, "Size")
i._shared_size = graceful_chain_get(i_data, "SharedSize")
i._virtual_size = graceful_chain_get(i_data, "VirtualSize")
return self._df
def extract_data_from_inspect(network_name, network_data):
"""
:param network_name: str
:param network_data: dict
:return: dict:
{
"ip_address4": "12.34.56.78"
"ip_address6": "ff:fa:..."
}
"""
a4 = None
if network_name == "host":
a4 = "127.0.0.1"
n = {}
a4 = graceful_chain_get(network_data, "IPAddress") or a4
if a4:
n["ip_address4"] = a4
a6 = graceful_chain_get(network_data, "GlobalIPv6Address")
if a6:
n["ip_address4"] = a6
return n