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

from shutil import copy, rmtree, copytree
from os.path import splitext, isfile, join, basename
from glob import glob

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

from tfw.config.logs import logging
log = logging.getLogger(__name__)


class FileManager:
    def __init__(self, source_directory, working_directory, selected_file=None):
        self._sourcedir = source_directory
        self._workdir = working_directory
        self._reload_files()
        self.filename = selected_file or basename(self.files[0])
        self.language = map_file_extension_to_language(self.filename)

    def select_file(self, filename):
        if not self._filepath(filename) in self.files:
           raise EnvironmentError('No such file in workdir!')
        self.filename = filename
        self.language = map_file_extension_to_language(self.filename)

    @property
    def files(self):
        return [file for file in glob(join(self._workdir, '**/*'), recursive=True) if isfile(file)]

    @property
    def file_contents(self):
        with open(self._filepath(self.filename), 'r') as ifile:
            return ifile.read()

    @file_contents.setter
    def file_contents(self, value):
        with open(self._filepath(self.filename), 'w') as ofile:
            ofile.write(value)

    def _filepath(self, filename):
        return join(self._workdir, filename)

    def _reload_files(self):
        rmtree(self._workdir, ignore_errors=True)
        copytree(self._sourcedir, self._workdir)


class SourceCodeEventHandler(EventHandlerBase, SupervisorMixin):
    def __init__(self, anchor, directory, process_name=None, selected_file=None):
        super().__init__(anchor)
        self.filemanager = FileManager(directory, LOGIN_APP_DIR, selected_file=selected_file)
        self.process_name = process_name

        self.commands = {
            'read': self.read,
            'write': self.write,
            'select': self.select
        }

        # Supervisor needs these to run the login program
        copy('source_code_server/server.py', LOGIN_APP_DIR)
        copy('source_code_server/users.db', LOGIN_APP_DIR)

    def read(self, data_json):
        data_json['data'] = {
            'filename': self.filemanager.filename,
            'content': self.filemanager.file_contents,
            'language': self.filemanager.language
        }

    def write(self, data_json):
        self.filemanager.file_contents = data_json['data']['content']
        self.restart_process()

    def select(self, data_json):
        try:
            self.filemanager.select_file(data_json['data']['filename'])
        except EnvironmentError:
            log.critical('Failed to select file "{}"'.format(data_json['data']['filename']))

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


def map_file_extension_to_language(filename):
    language_map = {
        # TODO: extend, maybe auto-generate???
        '.py': 'python',
        '.js': 'javascript'
    }
    _, extension = splitext(filename)
    language = 'javascript'
    try: language = language_map[extension]
    except KeyError: log.debug('No such extension in list, falling back to default: "{}".'.format(language))
    return language