Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def keypress(self, size, key):
logger.debug("size %r, key %r", size, key)
if key == "i":
docker_object = self.get_focus()[0].original_widget.docker_object
if isinstance(docker_object, DockerImage):
inspect_data = self.d.inspect_image(docker_object.image_id)
elif isinstance(docker_object, DockerContainer):
inspect_data = self.d.inspect_container(docker_object.container_id)
else:
raise RuntimeError("wat")
rendered_json = json.dumps(inspect_data, indent=2)
self.ui.add_and_set_main_widget(ScrollableListBox(rendered_json))
return
if key == "l":
docker_object = self.get_focus()[0].original_widget.docker_object
if isinstance(docker_object, DockerContainer):
logs_data, logs_generator = self.d.logs(docker_object.container_id)
w = AsyncScrollableListBox(logs_data, logs_generator, self.ui)
self.ui.add_and_set_main_widget(w)
return
key = super(MainListBox, self).keypress(size, key)
def get_images(self, cached=True):
if cached is False or self._images is None:
logger.debug("doing images() query")
self._images = {}
images_response = repeater(self.client.images) or []
for i in images_response:
img = DockerImage(i, self)
self._images[img.image_id] = img
self._all_images = {}
# FIXME: performance: do just all=True
all_images_response = repeater(self.client.images, kwargs={"all": True}) or []
for i in all_images_response:
img = DockerImage(i, self)
self._all_images[img.image_id] = img
return list(self._images.values())
def _assemble_initial_content(self):
widgets = []
for o in self.d.initial_content():
line = None
if isinstance(o, DockerImage):
image_id = urwid.Text(("image_id", o.image_id[:12]), align="left", wrap="any")
time = urwid.Text(("image_id", o.display_time_created()), align="left", wrap="any")
names = urwid.Text(("image_names", o.names or ""), align="left", wrap="clip")
line = DockerImageColumns(o, [(13, image_id), names, time])
elif isinstance(o, DockerContainer):
container_id = urwid.Text(("image_id", o.container_id[:12]), align="left", wrap="any")
time = urwid.Text(("image_id", o.display_time_created()), align="left", wrap="any")
name = urwid.Text(("image_names", o.name), align="left", wrap="clip")
command = urwid.Text(("image_names", o.command), align="left", wrap="clip")
status = urwid.Text(("image_names", o.status), align="left", wrap="clip")
line = DockerImageColumns(o, [(13, container_id), command, time, status, name])
widgets.append(urwid.AttrMap(line, 'image_id', focus_map='reversed'))
return widgets
def matches_search(self, s):
return s in self.image_id or any([s in str(x) for x in self.names])
def __str__(self):
# it's dangerous to put many stuff here b/c most of the values are loaded dynamically
# and it's trivial to go into nested exception madness
return "image {}".format(self.short_id)
def __repr__(self):
return self.__str__()
def containers(self):
return self.docker_backend.get_containers_for_image(self.image_id)
class RootImage(DockerImage):
"""
this is essentially "scratch" but you cannot inspect it anymore
"""
def __init__(self, docker_backend):
self.image_name = "scratch"
super().__init__(None, docker_backend, object_id="")
@property
def parent_id(self):
return None
@property
def parent_image(self):
return None
def get_next_sibling(self):
def run(self):
# TODO: needs better abstraction, backend object should be tied together with frontend
# object via a new Class
if isinstance(self.docker_object, DockerImage):
buffer_class = ImageInfoBuffer
elif isinstance(self.docker_object, DockerContainer):
buffer_class = ContainerInfoBuffer
else:
self.ui.notify_message("Can't display info for '%s'" % self.docker_object,
level="error")
logger.error("unable to display info buffer for %r", self.docker_object)
return
try:
# FIXME: this try/except block should be in upper frame
self.ui.add_and_display_buffer(buffer_class(self.docker_object, self.ui))
except NotifyError as ex:
self.ui.notify_message(str(ex), level="error")
logger.error(repr(ex))
def get_images(self, cached=True):
if cached is False or self._images is None:
logger.debug("doing images() query")
self._images = {}
images_response = repeater(self.client.images) or []
for i in images_response:
img = DockerImage(i, self)
self._images[img.image_id] = img
self._all_images = {}
# FIXME: performance: do just all=True
all_images_response = repeater(self.client.images, kwargs={"all": True}) or []
for i in all_images_response:
img = DockerImage(i, self)
self._all_images[img.image_id] = img
return list(self._images.values())
# "Tags": [
# "docker.io/fedora:latest"
# ],
# "Comment": "",
# "CreatedBy": "/bin/sh -c #(nop) ADD file:bcb5e5c... in /"
# }
try:
response = self.d.history(self.image_id)
except docker.errors.NotFound:
raise NotAvailableAnymore()
layers = []
for l in response:
layer_id = l["Id"]
if layer_id == "":
layers.append(DockerImage(l, self.docker_backend))
else:
layers.append(self.docker_backend.get_image_by_id(layer_id))
return layers
def get_row(docker_object):
if isinstance(docker_object, DockerImage):
return get_detailed_image_row(docker_object)
elif isinstance(docker_object, DockerContainer):
return get_detailed_container_row(docker_object)
else:
raise Exception("what ")