execute a python script line by line from another python script, using threads and bdb

line_by_line.py

import sys
import bdb
import threading

class ActualRunner(bdb.Bdb):
    def __init__(self):
        bdb.Bdb.__init__(self)
        self.line_no = -1
        self.program_done = False

    # Override Bdb methods
    def user_call(self, frame, argument_list):
        """Called if we might stop in a function."""
        # If the function being called is from the script, keep going line by line
        if frame.f_code.co_filename == "<string>":
            self.set_step()
        else:
            self.set_return(frame)

    def user_line(self, frame):
        """Called when we stop or break at a line."""
        self.line_end_event.set()
        self.run_line_event.wait()
        self.run_line_event.clear()
        if self.stop_thread:
            raise bdb.BdbQuit
        else:
            self.set_step()
            self.line_no = frame.f_lineno

    def user_return(self, frame, return_value):
        """Called when a return trap is set here."""
        name = frame.f_code.co_name or "<unknown>"
        if name == "<module>":
            self.program_done = True
            raise bdb.BdbQuit
        self.set_step()

    def user_exception(self, frame, exc_info):
        """Called when we stop on an exception."""
        raise exc_info[1]

class ProgramRunner(ActualRunner):
    def __init__(self, source: str):
        self.source = source
        self.exc_info = None

        ActualRunner.__init__(self)

        self.stop_thread = False
        self.run_line_event = threading.Event()
        self.line_end_event = threading.Event()
        self.run_thread = threading.Thread(target=self.thread_target)
        self.run_thread.start()

    def thread_target(self):
        try:
            self.run(self.source)
        except Exception as e:
            self.exc_info = sys.exc_info()
        finally:
            self.line_end_event.set()

    # This will raise the same exceptions that occured in the script being run
    # and will always wait until the line has been run before returning
    def update(self):
        self.line_end_event.clear()
        self.run_line_event.set()
        self.line_end_event.wait()
        self.line_end_event.clear()
        if self.exc_info:
            raise self.exc_info[1].with_traceback(self.exc_info[2])

    def stop(self):
        self.stop_thread = True
        self.run_line_event.set()
        self.run_thread.join()