#!/usr/bin/env python3
+"""Utilities for dealing with threads + threading."""
+
import functools
import logging
import os
def background_thread(
_funct: Optional[Callable],
-) -> Tuple[threading.Thread, threading.Event]:
+) -> 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
if _funct is None:
- return wrapper
+ return wrapper # type: ignore
else:
return wrapper(_funct)
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