Ran black code formatter on everything.

[python_utils.git] / thread_utils.py

diff --git a/thread_utils.py b/thread_utils.py
index af6e0e1abe840946a8754e426ca93bd653545ca5..ad1f0bf9029b3232ba9cbd28b085afade91f0186 100644 (file)
--- a/thread_utils.py
+++ b/thread_utils.py
@@ -6,6 +6,9 @@ import os
 import threading
 from typing import Callable, Optional, Tuple
 
+# This module is commonly used by others in here and should avoid
+# taking any unnecessary dependencies back on them.
+
 logger = logging.getLogger(__name__)
 
 
@@ -17,11 +20,14 @@ def current_thread_id() -> str:
 
 
 def is_current_thread_main_thread() -> bool:
+    """Returns True is the current (calling) thread is the process' main
+    thread and False otherwise.
+    """
     return threading.current_thread() is threading.main_thread()
 
 
 def background_thread(
-        _funct: Optional[Callable]
+    _funct: Optional[Callable],
 ) -> Tuple[threading.Thread, threading.Event]:
     """A function decorator to create a background thread.
 
@@ -52,10 +58,11 @@ def background_thread(
     periodically check.  If the event is set, it means the thread has
     been requested to terminate ASAP.
     """
+
     def wrapper(funct: Callable):
         @functools.wraps(funct)
         def inner_wrapper(
-                *a, **kwa
+            *a, **kwa
         ) -> Tuple[threading.Thread, threading.Event]:
             should_terminate = threading.Event()
             should_terminate.clear()
@@ -66,10 +73,9 @@ def background_thread(
                 kwargs=kwa,
             )
             thread.start()
-            logger.debug(
-                f'Started thread {thread.name} tid={thread.ident}'
-            )
+            logger.debug(f'Started thread {thread.name} tid={thread.ident}')
             return (thread, should_terminate)
+
         return inner_wrapper
 
     if _funct is None:
@@ -79,8 +85,8 @@ def background_thread(
 
 
 def periodically_invoke(
-        period_sec: float,
-        stop_after: Optional[int],
+    period_sec: float,
+    stop_after: Optional[int],
 ):
     """
     Periodically invoke a decorated function.  Stop after N invocations
@@ -102,6 +108,7 @@ def periodically_invoke(
             print(f"Hello, {name}")
 
     """
+
     def decorator_repeat(func):
         def helper_thread(should_terminate, *args, **kwargs) -> None:
             if stop_after is None:
@@ -124,14 +131,12 @@ def periodically_invoke(
             should_terminate.clear()
             newargs = (should_terminate, *args)
             thread = threading.Thread(
-                target=helper_thread,
-                args = newargs,
-                kwargs = kwargs
+                target=helper_thread, args=newargs, kwargs=kwargs
             )
             thread.start()
-            logger.debug(
-                f'Started thread {thread.name} tid={thread.ident}'
-            )
+            logger.debug(f'Started thread {thread.name} tid={thread.ident}')
             return (thread, should_terminate)
+
         return wrapper_repeat
+
     return decorator_repeat