Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for query_str in query_list:
if not query_str:
continue
# process here x=y queries and pass rest to parent filter()
try:
query_key, query_value = query_str.split("=", 1)
except ValueError:
unprocessed.append(query_str)
else:
logger.debug("looking up query key %r and query value %r", query_key, query_value)
for c in query_conf:
if query_key in c["query_keys"] and query_value in c["query_values"]:
c["callback"]()
break
else:
raise NotifyError("Invalid query string: %r", query_str)
widgets = []
logger.debug("doing query %s", backend_query)
query, c_op, i_op = self.d.filter(**backend_query)
for o in query:
try:
line = MainLineWidget(o)
except NotAvailableAnymore:
continue
widgets.append(line)
if unprocessed:
new_query = " ".join(unprocessed)
logger.debug("doing parent query for unprocessed string: %r", new_query)
super().filter(new_query, widgets_to_filter=widgets)
else:
def realtime_updates(self):
"""
fetch realtime events from docker and pass them to buffers
:return: None
"""
# TODO: make this available for every buffer
logger.info("starting receiving events from docker")
it = self.d.realtime_updates()
while True:
try:
event = next(it)
except NotifyError as ex:
self.ui.notify_message("error when receiving realtime events from docker: %s" % ex,
level="error")
return
# FIXME: we should pass events to all buffers
# ATM the buffers can't be rendered since they are not displayed
# and hence traceback like this: ListBoxError("Listbox contents too short! ...
logger.debug("pass event to current buffer %s", self.ui.current_buffer)
try:
self.ui.current_buffer.process_realtime_event(event)
except Exception as ex:
# swallow any exc
logger.error("error while processing runtime event: %r", ex)
def _search(self, reverse_search=False):
if self.search_string is None:
raise NotifyError("No search pattern specified.")
if not self.search_string:
self.search_string = None
return
pos = self.focus_position
original_position = pos
wrapped = False
while True:
if reverse_search:
obj, pos = self.body.get_prev(pos)
else:
obj, pos = self.body.get_next(pos)
if obj is None:
# wrap
wrapped = True
if reverse_search:
obj, pos = self.body[-1], len(self.body)
def __init__(self, docker_image, ui):
"""
:param docker_image:
:param ui: ui object so we refresh
"""
if isinstance(docker_image, RootImage):
raise NotifyError("Image \"scratch\" doesn't provide any more information.")
if docker_image.image_id == "":
raise NotifyError("This image (layer) is not available due to changes in docker-1.10 "
"image representation.")
self.docker_image = docker_image
self.display_name = docker_image.short_name
self.widget = ImageInfoWidget(ui, docker_image)
super().__init__()
def realtime_updates(self):
"""
update listing realtime as events come from docker
:return:
"""
# TODO: make this available for every buffer
logger.info("starting receiving events from docker")
it = self.d.realtime_updates()
while True:
try:
content = next(it)
except NotifyError as ex:
self.ui.notify_message(ex.args[0], level="error")
return
with self.realtime_lock:
if self.stop_realtime_events.is_set():
logger.info("received docker event when this functionality is disabled")
return
content.sort(key=attrgetter("natural_sort_value"), reverse=True)
widgets = []
for o in content:
try:
line = MainLineWidget(o)
except Exception as ex:
logger.error("%r", ex)
continue
if reverse_search:
obj, pos = self.body.get_prev(pos)
else:
obj, pos = self.body.get_next(pos)
if obj is None:
# wrap
wrapped = True
if reverse_search:
obj, pos = self.body[-1], len(self.body)
else:
obj, pos = self.body[0], 0
if wrapped and (
(pos > original_position and not reverse_search) or
(pos < original_position and reverse_search)
):
raise NotifyError("Pattern not found: %r." % self.search_string)
# FIXME: figure out nicer search api
if hasattr(obj, "matches_search"):
condition = obj.matches_search(self.search_string)
else:
if hasattr(obj, "original_widget"):
text = obj.original_widget.text
else:
text = obj.text
condition = self.search_string in text
if condition:
self.set_focus(pos)
self.reload_widget()
break
pre_message = "Getting logs for container {}...".format(docker_object.short_name)
ui.notify_message(pre_message)
if follow:
# FIXME: this is a bit race-y -- we might lose some logs with this approach
operation = docker_object.logs(follow=follow, lines=0)
static_data = docker_object.logs(follow=False).response
self.widget = AsyncScrollableListBox(operation.response, ui, static_data=static_data)
else:
operation = docker_object.logs(follow=follow)
self.widget = ScrollableListBox(ui, operation.response)
ui.remove_notification_message(pre_message)
ui.notify_widget(get_operation_notify_widget(operation, display_always=False))
except Exception as ex:
# FIXME: let's catch 404 and print that container doesn't exist
# instead of printing ugly HTTP error
raise NotifyError("Error getting logs for container %s: %r" % (docker_object, ex))
else:
raise NotifyError("Only containers have logs.")
super().__init__()
def filter(ui, oldfooter, edit_widget, text_input):
logger.debug("%r %r", edit_widget, text_input)
if text_input.endswith("\n"):
ui.mainframe.prompt_bar = None
ui.mainframe.set_footer(oldfooter)
try:
ui.current_buffer.filter(text_input[:-1])
except NotifyError as ex:
logger.error(repr(ex))
ui.notify_message(str(ex), level="error")
ui.mainframe.set_focus("body")
ui.reload_footer()