Function worker

pyhooks/pyhooks/python_server.py:171–199

Redirects outputs and performs exec

(code: str, output_file, timeout_fyi: float, log: bool)

def worker(code: str, output_file, timeout_fyi: float, log: bool):
    "Redirects outputs and performs exec"

    # set up output redirection
    c = worker_counter_box[0]
    worker_counter_box[0] += 1
    stdouts: list = [output_file]
    stderrs: list = [output_file]
    if log:
        stdouts.append(PrefixedFile(real_stdout, f"[python-exec-{c}]-  "))
        stderrs.append(PrefixedFile(real_stderr, f"[python-exec-{c}]+  "))
    stdout.set_outputs(stdouts)
    stderr.set_outputs(stderrs)

    # do the exec
    try:
        ipython_shell.run_cell(code)
    except PythonExecTimeoutException:
        print(
            f"PythonExecTimeoutException: python exec timed out after {timeout_fyi} seconds",
            file=stderr,
        )
    except PythonExecOutOfMemoryException:
        print(
            "PythonExecOutOfMemoryException: python exec exceeded available memory. Python environment has been reset.",
            file=stderr,
        )
    except Exception as e:
        traceback.print_exception(type(e), e, e.__traceback__, file=stderr)


def python_exec_sync(