Created
October 14, 2018 00:41
-
-
Save LiquidFenrir/186a8c96c987b9e896f35c99e2ba9b57 to your computer and use it in GitHub Desktop.
execute a python script line by line from another python script, using threads and bdb
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment