Skip to content

Instantly share code, notes, and snippets.

@balmacefa
Created August 24, 2021 20:57
Show Gist options
  • Select an option

  • Save balmacefa/20f2384c57a0d39a2d8c5909ed6e0e7e to your computer and use it in GitHub Desktop.

Select an option

Save balmacefa/20f2384c57a0d39a2d8c5909ed6e0e7e to your computer and use it in GitHub Desktop.
Python Thread Class, catch any exception in the Thread and raise to the caller
import sys
import threading
# Custom Thread Class
# catch any exception in the Thread and raise to the caller
class ExceptionThread(threading.Thread):
def __init__(self, *args, **kwargs):
super(ExceptionThread, self).__init__(*args, **kwargs)
self._exc = None
def run(self):
# Variable that stores the exception, if raised by someFunction
self._exc = None
try:
if self._target:
self._target(*self._args, **self._kwargs)
except BaseException as e:
self._exc = e
def join(self, timeout=None):
threading.Thread.join(self)
# Since join() returns in caller thread
# we re-raise the caught exception
# if any was caught
if self._exc:
raise self._exc
def main():
run_event = threading.Event()
run_event.set()
def start_thread(id, _run_event):
raise RuntimeError(f'Thread id: {id}')
# start a new Thread
thread = ExceptionThread(target=start_thread, args=(id, run_event))
try:
thread.start()
# this raise the exceptions in the Thread
thread.join()
except RuntimeError as error:
print(error)
return 0
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment