baseimage-tutorial-framework/src/event_handlers/source_code_event_handler.py

from shutil import copy, rmtree, copytree
from os.path import splitext
from contextlib import suppress
from xmlrpc.client import Fault as SupervisorFault

from tfw.util import SupervisorMixin
from tfw.config import LOGIN_APP_DIR
from tfw.event_handler_base import EventHandlerBase


class SourceCodeEventHandler(EventHandlerBase, SupervisorMixin):
    def __init__(self, anchor, filename, process_name=None):
        super().__init__(anchor)
        self.working_directory = LOGIN_APP_DIR
        self.filename = filename
        self.language = map_file_extension_to_language(filename)
        self.process_name = process_name or splitext(filename)[0]
        self.commands = {
            'read': self.read_file,
            'write': self.write_file
        }

        self.file = self.create_initial_state()

    def handle_event(self, anchor, data_json):
        data = data_json['data']
        self.commands[data['command']](data_json)
        return data_json

    def handle_reset(self, data_json):
        self.create_initial_state()

    def read_file(self, data_json):
        with open(self.file, 'r') as ifile:
            content = ifile.read()
        data_json['data'] = {
            'filename': self.filename,
            'content': content,
            'language': self.language
        }

    def write_file(self, data_json):
        data = data_json['data']
        with open(self.file, 'w') as ofile:
            ofile.write(data['content'])
        self.supervisor.stopProcess(self.process_name)
        self.supervisor.startProcess(self.process_name)

    def create_initial_state(self):
        with suppress(SupervisorFault):
            self.supervisor.stopProcess(self.process_name)

        rmtree(self.working_directory, ignore_errors=True)
        copytree('source_code_server/', self.working_directory)
        file = copy(self.filename, self.working_directory)

        self.supervisor.startProcess(self.process_name)

        return file


def map_file_extension_to_language(filename):
    language_map = {
        # TODO: extend, maybe auto-generate???
        '.py': 'python',
        '.js': 'javascript'
    }
    _, extension = splitext(filename)
    return language_map[extension]