最初はsys.excepthook
、解決策のように見えました。これは、キャッチされない例外がスローされるたびに呼び出されるグローバル フックです。
残念ながら、これは機能しません。なんで?画面に表示される素敵なトレースバックを出力するコードで関数を適切にthreading
ラップrun
します (常にどのように表示されるかに注意してくださいException in thread {Name of your thread here}
。これがその方法です)。
Python 3.8 から、これを機能させるためにオーバーライドできる関数があります: threading.excepthook
... threading.excepthook() をオーバーライドして、Thread.run() によって発生したキャッチされていない例外の処理方法を制御できます
どうしようか?この関数をロジックに置き換えてください。
Python >= 3.8 の場合
import traceback
import threading
import os
class GlobalExceptionWatcher(object):
def _store_excepthook(self, args):
'''
Uses as an exception handlers which stores any uncaught exceptions.
'''
self.__org_hook(args)
formated_exc = traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback)
self._exceptions.append('\n'.join(formated_exc))
return formated_exc
def __enter__(self):
'''
Register us to the hook.
'''
self._exceptions = []
self.__org_hook = threading.excepthook
threading.excepthook = self._store_excepthook
def __exit__(self, type, value, traceback):
'''
Remove us from the hook, assure no exception were thrown.
'''
threading.excepthook = self.__org_hook
if len(self._exceptions) != 0:
tracebacks = os.linesep.join(self._exceptions)
raise Exception(f'Exceptions in other threads: {tracebacks}')
古いバージョンの Python の場合、これはもう少し複雑です。簡単に言えば、threading
結節には、次のような行に沿って何かを行う文書化されていないインポートがあるようです。
threading._format_exc = traceback.format_exc
驚くことではありませんが、この関数は、スレッドのrun
関数から例外がスローされたときにのみ呼び出されます。
したがって、python <= 3.7の場合
import threading
import os
class GlobalExceptionWatcher(object):
def _store_excepthook(self):
'''
Uses as an exception handlers which stores any uncaught exceptions.
'''
formated_exc = self.__org_hook()
self._exceptions.append(formated_exc)
return formated_exc
def __enter__(self):
'''
Register us to the hook.
'''
self._exceptions = []
self.__org_hook = threading._format_exc
threading._format_exc = self._store_excepthook
def __exit__(self, type, value, traceback):
'''
Remove us from the hook, assure no exception were thrown.
'''
threading._format_exc = self.__org_hook
if len(self._exceptions) != 0:
tracebacks = os.linesep.join(self._exceptions)
raise Exception('Exceptions in other threads: %s' % tracebacks)
使用法:
my_thread = x.ExceptionRaiser()
# will fail when thread is started and raises an exception.
with GlobalExceptionWatcher():
my_thread.start()
my_thread.join()
それでもjoin
自分で行う必要がありますが、終了時に with ステートメントのコンテキスト マネージャーが他のスレッドでスローされた例外をチェックし、適切に例外を発生させます。
コードは「現状のまま」提供され、明示または黙示を問わず、いかなる種類の保証もありません
これは文書化されていない、一種の恐ろしいハックです。Linux と Windows でテストしましたが、動作するようです。ご自身の責任で使用してください。