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

64 lines
2.1 KiB
Python
Raw Normal View History

import json
from shutil import copy, rmtree, copytree
from os.path import splitext
import xmlrpc.client
from config import SUPERVISOR_HTTP_URI, LOGIN_APP_DIR
from event_handler_base import EventHandlerBase
class SourceCodeEventHandler(EventHandlerBase):
def __init__(self, anchor, filename, process_name=None, zmq_context=None):
super().__init__(anchor, zmq_context)
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]
server = xmlrpc.client.ServerProxy(SUPERVISOR_HTTP_URI)
self.supervisor = server.supervisor
self.file = self.create_initial_state()
def handle_event(self, anchor, data_json):
data = data_json['data']
if anchor == b'reset':
self.file = self.create_initial_state(process_is_running=True)
if data['command'] == 'read':
with open(self.file, 'r') as ifile:
content = ifile.read()
data_json['data'] = {
'filename': self.filename,
'content': content,
'language': self.language
}
return data_json
elif data['command'] == 'write':
with open(self.file, 'w') as ofile:
ofile.write(data['content'])
self.supervisor.stopProcess(self.process_name)
self.supervisor.startProcess(self.process_name)
return None
def create_initial_state(self, process_is_running=False):
if process_is_running:
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]