projects
/
python_utils.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Add LICENSE and NOTICE.
[python_utils.git]
/
thread_utils.py
diff --git
a/thread_utils.py
b/thread_utils.py
index 6035b09ef2df448a7dba5ad46c90af2e594c5dbf..01755deafc1e7af9189026c2ae233c6146c7d494 100644
(file)
--- a/
thread_utils.py
+++ b/
thread_utils.py
@@
-1,10
+1,14
@@
#!/usr/bin/env python3
#!/usr/bin/env python3
+# © Copyright 2021-2022, Scott Gasch
+
+"""Utilities for dealing with threads + threading."""
+
import functools
import logging
import os
import 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.
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
@@
-60,8
+64,8
@@
def is_current_thread_main_thread() -> bool:
def background_thread(
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 ***
"""A function decorator to create a background thread.
*** Please note: the decorated function must take an shutdown ***
@@
-104,7
+108,7
@@
def background_thread(
kwargs=kwa,
)
thread.start()
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
return (thread, should_terminate)
return inner_wrapper
@@
-163,7
+167,7
@@
def periodically_invoke(
newargs = (should_terminate, *args)
thread = threading.Thread(target=helper_thread, args=newargs, kwargs=kwargs)
thread.start()
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
return (thread, should_terminate)
return wrapper_repeat