#!/usr/bin/env python3
+# © Copyright 2021-2022, Scott Gasch
+
+"""Utilities for dealing with threads + threading."""
+
import functools
import logging
import os
import threading
-from typing import Callable, Optional, Tuple
+from typing import Any, Callable, Optional, Tuple
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
def background_thread(
- _funct: Optional[Callable],
-) -> Tuple[threading.Thread, threading.Event]:
+ _funct: Optional[Callable[..., Any]],
+) -> Callable[..., Tuple[threading.Thread, threading.Event]]:
"""A function decorator to create a background thread.
*** Please note: the decorated function must take an shutdown ***
kwargs=kwa,
)
thread.start()
- logger.debug(f'Started thread {thread.name} tid={thread.ident}')
+ logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
return (thread, should_terminate)
return inner_wrapper
newargs = (should_terminate, *args)
thread = threading.Thread(target=helper_thread, args=newargs, kwargs=kwargs)
thread.start()
- logger.debug(f'Started thread {thread.name} tid={thread.ident}')
+ logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
return (thread, should_terminate)
return wrapper_repeat