iframe-proxy | Sunbelt Computer Software

120 lines (109 loc) · 4.12 KB
import json
import base64
import traceback
import inspect
class Handler(object):
    def __init__(self, apply_func, load_func):
        """
        Creates the handler object
        :param apply_func: A required function that can have an arity of 1-2, depending on if loading occurs
        :param load_func: An optional supplier function used if load time events are required, has an arity of 0.
        """
        self.FIFO_PATH = "/tmp/algoout"
        apply_args, _, _, apply_defaults = inspect.getargspec(apply_func)
        load_args, _, _, _ = inspect.getargspec(load_func)
        if len(load_args) > 0:
            raise Exception("load function must not have parameters")
        if len(apply_args) > 2 or len(apply_args) == 0:
            raise Exception("apply function may have between 1 and 2 parameters, not {}".format(len(apply_args)))
        self.apply_func = apply_func
        self.load_func = load_func
        self.load_result = None
    def load(self):
        self.load_result = self.load_func()
        print('PIPE_INIT_COMPLETE')
        sys.stdout.flush()
    def format_data(self, request):
        if request['content_type'] in ['text', 'json']:
            data = request['data']
        elif request['content_type'] == 'binary':
            data = self.wrap_binary_data(request['data'])
        else:
            raise Exception("Invalid content_type: {}".format(request['content_type']))
        return data
    def is_binary(self, arg):
        if six.PY3:
            return isinstance(arg, base64.bytes_types)
        return isinstance(arg, bytearray)
    def wrap_binary_data(self, data):
        if six.PY3:
            return bytes(data)
        else:
            return bytearray(data)
    def format_response(self, response):
        if self.is_binary(response):
            content_type = 'binary'
            response = base64.b64encode(response)
            if not isinstance(response, six.string_types):
                response = str(response, 'utf-8')
        elif isinstance(response, six.string_types) or isinstance(response, six.text_type):
            content_type = 'text'
        else:
            content_type = 'json'
        response_string = json.dumps({
            'result': response,
            'metadata': {
                'content_type': content_type
        return response_string
    def write_to_pipe(self, data_string):
        if os.name == "posix":
            with open(self.FIFO_PATH, 'w') as f:
                f.write(data_string)
                f.write('\n')
            sys.stdout.flush()
        if os.name == "nt":
            sys.stdin = data_string
    def serve(self):
        try:
            self.load()
        except Exception as e:
            if hasattr(e, 'error_type'):
                error_type = e.error_type
            else:
                error_type = 'AlgorithmError'
            load_error_string = json.dumps({
                'error': {
                    'message': str(e),
                    'stacktrace': traceback.format_exc(),
                    'error_type': error_type
            return self.write_to_pipe(load_error_string)
        for line in sys.stdin:
            try:
                request = json.loads(line)
                formatted_input = self.format_data(request)
                if self.load_result:
                    apply_result = self.apply_func(formatted_input, self.load_result)
                else:
                    apply_result = self.apply_func(formatted_input)
                response_string = self.format_response(apply_result)
            except Exception as e:
                if hasattr(e, 'error_type'):
                    error_type = e.error_type
                else:
                    error_type = 'AlgorithmError'
                response_string = json.dumps({
                    'error': {
                        'message': str(e),
                        'stacktrace': traceback.format_exc(),
                        'error_type': error_type
            finally:
                self.write_to_pipe(response_string)
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

handler.py

Latest commit

History

handler.py

File metadata and controls
