Skip to content

Instantly share code, notes, and snippets.

@mrmamongo
Created September 20, 2023 06:26
Show Gist options
  • Save mrmamongo/bb835c4a7ff975405a25d1ecbc5e068d to your computer and use it in GitHub Desktop.
Save mrmamongo/bb835c4a7ff975405a25d1ecbc5e068d to your computer and use it in GitHub Desktop.
Convenience class for stoppable threads
class BaseThread(threading.Thread):
def __init__(self):
threading.Thread.init(self)
if hasattr(self, "daemon"):
self.daemon = True
else:
self.setDaemon(True)
self._stopped_event = threading.Event()
@property
def stopped_event(self):
return self._stopped_event
def should_keep_running(self):
"""Determines whether the thread should continue running."""
return not self._stopped_event.is_set()
def on_thread_stop(self):
"""Override this method instead of :meth:`stop()`.
:meth:`stop()` calls this method.
This method is called immediately after the thread is signaled to stop.
"""
pass
def stop(self):
"""Signals the thread to stop."""
self._stopped_event.set()
self.on_thread_stop()
def on_thread_start(self):
"""Override this method instead of :meth:`start()`. :meth:`start()`
calls this method.
This method is called right before this thread is started and this
object’s run() method is invoked.
"""
pass
def start(self):
self.on_thread_start()
threading.Thread.start(self)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment