99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
from typing import Dict, Union
|
|
|
|
from asciimatics.widgets import Label, Button, Layout, Frame
|
|
import requests
|
|
|
|
from .Model import Model
|
|
from ..StreamResponseHandler import StreamResponseHandler
|
|
|
|
|
|
class TUIStreamResponseHandler(StreamResponseHandler):
|
|
def __init__(
|
|
self,
|
|
model: Model,
|
|
labels: Dict[str, Label],
|
|
next_btn: Button,
|
|
msg_layout: Layout,
|
|
frame: Frame,
|
|
):
|
|
self.screen = model.screen
|
|
self.operations = model.operations
|
|
self.idx = 0
|
|
self.labels = labels
|
|
self.next_btn = next_btn
|
|
self.msg_layout = msg_layout
|
|
self.frame = frame
|
|
self.error_messages = []
|
|
|
|
def _update(self):
|
|
# Since we're running in a separate thread, we need to force the
|
|
# screen to update. See
|
|
# https://github.com/peterbrittain/asciimatics/issues/56
|
|
self.frame.fix()
|
|
self.screen.force_update()
|
|
|
|
def _enable_next_btn(self):
|
|
self.next_btn.disabled = False
|
|
self.frame.reset()
|
|
|
|
def _show_msg(self, msg: str = ''):
|
|
for line in msg.splitlines():
|
|
self.msg_layout.add_widget(Label(line, align='^'))
|
|
|
|
def _abort(self):
|
|
for operation in self.operations[self.idx:]:
|
|
self.labels[operation].text = 'ABORTED'
|
|
self._enable_next_btn()
|
|
|
|
def handle_non_200(self, resp: requests.Response):
|
|
self._abort()
|
|
self._show_msg('An error occurred:')
|
|
self._show_msg(resp.text)
|
|
self._update()
|
|
|
|
def begin(self):
|
|
pass
|
|
|
|
def handle_aborted(self, err_msg: str):
|
|
self._abort()
|
|
self._show_msg('The transaction was rolled back.')
|
|
self._show_msg('The error was:')
|
|
self._show_msg(err_msg)
|
|
self._show_msg('Please check the ceod logs.')
|
|
self._update()
|
|
|
|
def handle_completed(self):
|
|
self._show_msg('Transaction successfully completed.')
|
|
if len(self.error_messages) > 0:
|
|
self._show_msg('There were some errors, please check the '
|
|
'ceod logs.')
|
|
# we don't have enough space in the TUI to actually
|
|
# show the error messages
|
|
self._enable_next_btn()
|
|
self._update()
|
|
|
|
def handle_successful_operation(self):
|
|
operation = self.operations[self.idx]
|
|
self.labels[operation].text = 'Done'
|
|
self.idx += 1
|
|
self._update()
|
|
|
|
def handle_failed_operation(self, err_msg: Union[str, None]):
|
|
operation = self.operations[self.idx]
|
|
self.labels[operation].text = 'Failed'
|
|
if err_msg is not None:
|
|
self.error_messages.append(err_msg)
|
|
self.idx += 1
|
|
self._update()
|
|
|
|
def handle_skipped_operation(self):
|
|
operation = self.operations[self.idx]
|
|
self.labels[operation].text = 'Skipped'
|
|
self.idx += 1
|
|
self._update()
|
|
|
|
def handle_unrecognized_operation(self, operation: str):
|
|
self.error_messages.append('Unrecognized operation: ' + operation)
|
|
self.idx += 1
|
|
self._update()
|