pyceo/ceo/tui/TUIStreamResponseHandler.py

from typing import Dict, Union

from asciimatics.widgets import Label, Layout
import requests

from .Model import Model
from ..StreamResponseHandler import StreamResponseHandler


class TUIStreamResponseHandler(StreamResponseHandler):
    def __init__(
        self,
        model: Model,
        labels: Dict[str, Label],
        msg_layout: Layout,
        txn_view,  # TransactionView
    ):
        super().__init__()
        self.screen = model.screen
        self.operations = model.operations
        self.idx = 0
        self.labels = labels
        self.msg_layout = msg_layout
        self.txn_view = txn_view
        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.txn_view.fix()
        self.screen.force_update()

    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.txn_view.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:')
            for msg in self.error_messages:
                self._show_msg(msg)
        self.txn_view.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()