Attraper l'exception d'un thread dans le thread appelant en Python

Bien qu'il ne soit pas possible d'attraper directement une exception lancée dans un thread différent, voici un code pour obtenir de manière assez transparente quelque chose de très proche de cette fonctionnalité. Votre thread enfant doit sous-classer la classe ExThread au lieu de threading.Thread et le thread parent doit appeler la méthode child_thread.join_with_exception() au lieu de child_thread.join() en attendant que le thread termine son travail.

Détails techniques de cette implémentation: lorsque le thread enfant lève une exception, il est passé au parent via un {[5] } et jeté à nouveau dans le thread parent. Notez qu'il n'y a pas d'attente occupée dans cette approche .

#!/usr/bin/env python

import sys
import threading
import Queue

class ExThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.__status_queue = Queue.Queue()

    def run_with_exception(self):
        """This method should be overriden."""
        raise NotImplementedError

    def run(self):
        """This method should NOT be overriden."""
        try:
            self.run_with_exception()
        except BaseException:
            self.__status_queue.put(sys.exc_info())
        self.__status_queue.put(None)

    def wait_for_exc_info(self):
        return self.__status_queue.get()

    def join_with_exception(self):
        ex_info = self.wait_for_exc_info()
        if ex_info is None:
            return
        else:
            raise ex_info[1]

class MyException(Exception):
    pass

class MyThread(ExThread):
    def __init__(self):
        ExThread.__init__(self)

    def run_with_exception(self):
        thread_name = threading.current_thread().name
        raise MyException("An error in thread '{}'.".format(thread_name))

def main():
    t = MyThread()
    t.start()
    try:
        t.join_with_exception()
    except MyException as ex:
        thread_name = threading.current_thread().name
        print "Caught a MyException in thread '{}': {}".format(thread_name, ex)

if __name__ == '__main__':
    main()

Le concurrent.futures le module facilite le travail dans des threads (ou des processus) séparés et gère les exceptions résultantes:

import concurrent.futures
import shutil

def copytree_with_dots(src_path, dst_path):
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        # Execute the copy on a separate thread,
        # creating a future object to track progress.
        future = executor.submit(shutil.copytree, src_path, dst_path)

        while future.running():
            # Print pretty dots here.
            pass

        # Return the value returned by shutil.copytree(), None.
        # Raise any exceptions raised during the copy process.
        return future.result()

concurrent.futures est inclus avec Python 3.2, et est disponible en tant que le module rétroporté futures pour les versions antérieures.

Si une exception se produit dans un thread, le meilleur moyen est de la relancer dans le thread appelant pendant join. Vous pouvez obtenir des informations sur l'exception en cours de traitement à l'aide de la fonction sys.exc_info(). Ces informations peuvent simplement être stockées en tant que propriété de l'objet thread jusqu'à ce que join soit appelé, auquel cas elles peuvent être réactivées.

Notez qu'un Queue.Queue (comme suggéré dans d'autres réponses) n'est pas nécessaire dans ce cas simple où le thread lève au maximum 1 exception et se termine juste après avoir lancé une exception . Nous évitons les conditions de course en attendant simplement que le fil se termine.

Par exemple, étendre ExcThread (ci-dessous), en remplaçant excRun (au lieu de run).

Python 2.x:

import threading

class ExcThread(threading.Thread):
  def excRun(self):
    pass

  def run(self):
    self.exc = None
    try:
      # Possibly throws an exception
      self.excRun()
    except:
      import sys
      self.exc = sys.exc_info()
      # Save details of the exception thrown but don't rethrow,
      # just complete the function

  def join(self):
    threading.Thread.join(self)
    if self.exc:
      msg = "Thread '%s' threw an exception: %s" % (self.getName(), self.exc[1])
      new_exc = Exception(msg)
      raise new_exc.__class__, new_exc, self.exc[2]

Python 3.x:

Le formulaire 3 arguments pour raise est parti en Python 3, donc changez la dernière ligne en:

raise new_exc.with_traceback(self.exc[2])

Il y a beaucoup de réponses vraiment bizarrement compliquées à cette question. Est-ce que je simplifie trop, parce que cela me semble suffisant pour la plupart des choses.

from threading import Thread

class PropagatingThread(Thread):
    def run(self):
        self.exc = None
        try:
            if hasattr(self, '_Thread__target'):
                # Thread uses name mangling prior to Python 3.
                self.ret = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
            else:
                self.ret = self._target(*self._args, **self._kwargs)
        except BaseException as e:
            self.exc = e

    def join(self):
        super(PropagatingThread, self).join()
        if self.exc:
            raise self.exc
        return self.ret

Si vous êtes certain que vous ne courrez que sur l'une ou l'autre version de Python, vous pouvez réduire la méthode run() à la version mutilée (si vous ne courrez que sur les versions de Python avant 3), ou juste la version propre (si vous ne courrez que sur les versions de Python commençant par 3).

Exemple d'utilisation:

def f(*args, **kwargs)
    print(args)
    print(kwargs)
    raise Exception('I suck')

t = PropagatingThread(target=f, args=(5,), kwargs={'hello':'world'})
t.start()
t.join()

Et vous verrez l'exception soulevée sur l'autre fil lorsque vous vous joignez.

C'était un vilain petit problème, et j'aimerais lancer ma solution. D'autres solutions que j'ai trouvées (async.io par exemple) semblait prometteur, mais a également présenté un peu d'une boîte noire. L'approche de boucle de file d'attente / événement vous lie en quelque sorte à une certaine implémentation. le code source de futures simultanées, cependant, est autour de seulement 1000 lignes, et facile à comprendre . Cela m'a permis de résoudre facilement mon problème: créer des threads de travail ad-hoc sans trop de configuration, et pouvoir attraper des exceptions dans le thread principal.

Ma solution utilise l'API futures concomitante et L'API threading. Il vous permet de créer un worker qui vous donne à la fois le thread et le futur. De cette façon, vous pouvez joindre le fil attendre le résultat:

worker = Worker(test)
thread = worker.start()
thread.join()
print(worker.future.result())

...ou vous pouvez laisser le travailleur envoyer un rappel une fois terminé:

worker = Worker(test)
thread = worker.start(lambda x: print('callback', x))

...ou vous pouvez faire une boucle jusqu'à la fin de l'événement:

worker = Worker(test)
thread = worker.start()

while True:
    print("waiting")
    if worker.future.done():
        exc = worker.future.exception()
        print('exception?', exc)
        result = worker.future.result()
        print('result', result)           
        break
    time.sleep(0.25)

Voici le code:

from concurrent.futures import Future
import threading
import time

class Worker(object):
    def __init__(self, fn, args=()):
        self.future = Future()
        self._fn = fn
        self._args = args

    def start(self, cb=None):
        self._cb = cb
        self.future.set_running_or_notify_cancel()
        thread = threading.Thread(target=self.run, args=())
        thread.daemon = True #this will continue thread execution after the main thread runs out of code - you can still ctrl + c or kill the process
        thread.start()
        return thread

    def run(self):
        try:
            self.future.set_result(self._fn(*self._args))
        except BaseException as e:
            self.future.set_exception(e)

        if(self._cb):
            self._cb(self.future.result())

...et la fonction de test:

def test(*args):
    print('args are', args)
    time.sleep(2)
    raise Exception('foo')

En tant que Noobie to Threading, il m'a fallu beaucoup de temps pour comprendre comment implémenter le code de Mateusz Kobos (ci-dessus). Voici une version clarifiée pour aider à comprendre comment l'utiliser.

#!/usr/bin/env python

import sys
import threading
import Queue

class ExThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.__status_queue = Queue.Queue()

    def run_with_exception(self):
        """This method should be overriden."""
        raise NotImplementedError

    def run(self):
        """This method should NOT be overriden."""
        try:
            self.run_with_exception()
        except Exception:
            self.__status_queue.put(sys.exc_info())
        self.__status_queue.put(None)

    def wait_for_exc_info(self):
        return self.__status_queue.get()

    def join_with_exception(self):
        ex_info = self.wait_for_exc_info()
        if ex_info is None:
            return
        else:
            raise ex_info[1]

class MyException(Exception):
    pass

class MyThread(ExThread):
    def __init__(self):
        ExThread.__init__(self)

    # This overrides the "run_with_exception" from class "ExThread"
    # Note, this is where the actual thread to be run lives. The thread
    # to be run could also call a method or be passed in as an object
    def run_with_exception(self):
        # Code will function until the int
        print "sleeping 5 seconds"
        import time
        for i in 1, 2, 3, 4, 5:
            print i
            time.sleep(1) 
        # Thread should break here
        int("str")
# I'm honestly not sure why these appear here? So, I removed them. 
# Perhaps Mateusz can clarify?        
#         thread_name = threading.current_thread().name
#         raise MyException("An error in thread '{}'.".format(thread_name))

if __name__ == '__main__':
    # The code lives in MyThread in this example. So creating the MyThread 
    # object set the code to be run (but does not start it yet)
    t = MyThread()
    # This actually starts the thread
    t.start()
    print
    print ("Notice 't.start()' is considered to have completed, although" 
           " the countdown continues in its new thread. So you code "
           "can tinue into new processing.")
    # Now that the thread is running, the join allows for monitoring of it
    try:
        t.join_with_exception()
    # should be able to be replace "Exception" with specific error (untested)
    except Exception, e: 
        print
        print "Exceptioon was caught and control passed back to the main thread"
        print "Do some handling here...or raise a custom exception "
        thread_name = threading.current_thread().name
        e = ("Caught a MyException in thread: '" + 
             str(thread_name) + 
             "' [" + str(e) + "]")
        raise Exception(e) # Or custom class of exception, such as MyException

De manière similaire à RickardSjogren sans file d'attente, sys etc. mais aussi sans quelques écouteurs aux signaux: exécutez directement un gestionnaire d'exception qui correspond à un bloc except.

#!/usr/bin/env python3

import threading

class ExceptionThread(threading.Thread):

    def __init__(self, callback=None, *args, **kwargs):
        """
        Redirect exceptions of thread to an exception handler.

        :param callback: function to handle occured exception
        :type callback: function(thread, exception)
        :param args: arguments for threading.Thread()
        :type args: tuple
        :param kwargs: keyword arguments for threading.Thread()
        :type kwargs: dict
        """
        self._callback = callback
        super().__init__(*args, **kwargs)

    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        except BaseException as e:
            if self._callback is None:
                raise e
            else:
                self._callback(self, e)
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            del self._target, self._args, self._kwargs, self._callback

Seulement soi-même._callback et le bloc except dans run () est supplémentaire au threading normal.Fil.

Une méthode que j'aime est basée sur le modèle d'observateur . Je définis une classe de signal que mon thread utilise pour émettre des exceptions aux auditeurs. Il peut également être utilisé pour renvoyer des valeurs à partir de threads. Exemple:

import threading

class Signal:
    def __init__(self):
        self._subscribers = list()

    def emit(self, *args, **kwargs):
        for func in self._subscribers:
            func(*args, **kwargs)

    def connect(self, func):
        self._subscribers.append(func)

    def disconnect(self, func):
        try:
            self._subscribers.remove(func)
        except ValueError:
            raise ValueError('Function {0} not removed from {1}'.format(func, self))


class WorkerThread(threading.Thread):

    def __init__(self, *args, **kwargs):
        super(WorkerThread, self).__init__(*args, **kwargs)
        self.Exception = Signal()
        self.Result = Signal()

    def run(self):
        if self._Thread__target is not None:
            try:
                self._return_value = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
            except Exception as e:
                self.Exception.emit(e)
            else:
                self.Result.emit(self._return_value)

if __name__ == '__main__':
    import time

    def handle_exception(exc):
        print exc.message

    def handle_result(res):
        print res

    def a():
        time.sleep(1)
        raise IOError('a failed')

    def b():
        time.sleep(2)
        return 'b returns'

    t = WorkerThread(target=a)
    t2 = WorkerThread(target=b)
    t.Exception.connect(handle_exception)
    t2.Result.connect(handle_result)
    t.start()
    t2.start()

    print 'Threads started'

    t.join()
    t2.join()
    print 'Done'

Je n'ai pas assez d'expérience de travail avec des threads pour prétendre que c'est une méthode complètement sûre. Mais cela a fonctionné pour moi et j'aime la flexibilité.

Utiliser des excepts nus n'est pas une bonne pratique car vous attrapez généralement plus que vous ne négociez.

Je suggère de modifier le except pour attraper uniquement l'exception que vous souhaitez gérer. Je ne pense pas que l'élever ait l'effet désiré, car quand vous allez instancier TheThread dans l'extérieur try, s'il déclenche une exception, l'affectation ne se produira jamais.

Au lieu de cela, vous voudrez peut-être simplement alerter et passer à autre chose, comme:

def run(self):
    try:
       shul.copytree(self.sourceFolder, self.destFolder)
    except OSError, err:
       print err

Alors quand cette exception est interceptée, vous pouvez la gérer là-bas. Ensuite, lorsque l'extérieur try attrape une exception de TheThread, vous savez que ce ne sera pas celui que vous avez déjà géré, et vous aidera à isoler votre flux de processus.

Un moyen simple d'attraper l'exception du thread et de communiquer à la méthode de l'appelant pourrait être de passer dictionary ou une list à la méthode worker.

Exemple (passage du dictionnaire à la méthode worker):

import threading

def my_method(throw_me):
    raise Exception(throw_me)

def worker(shared_obj, *args, **kwargs):
    try:
        shared_obj['target'](*args, **kwargs)
    except Exception as err:
        shared_obj['err'] = err

shared_obj = {'err':'', 'target': my_method}
throw_me = "Test"

th = threading.Thread(target=worker, args=(shared_obj, throw_me), kwargs={})
th.start()
th.join()

if shared_obj['err']:
    print(">>%s" % shared_obj['err'])

Je sais que je suis un peu en retard à la fête ici, mais j'avais un problème très similaire, mais cela incluait l'utilisation de tkinter comme interface graphique, et le mainloop rendait impossible l'utilisation des solutions qui dépendent .rejoindre(). Par conséquent, j'ai adapté la solution donnée dans L'édition de la question originale, mais l'ai rendue plus générale pour la rendre plus facile à comprendre pour les autres.

Voici la nouvelle classe de threads en action:

import threading
import traceback
import logging


class ExceptionThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)

    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        except Exception:
            logging.error(traceback.format_exc())


def test_function_1(input):
    raise IndexError(input)


if __name__ == "__main__":
    input = 'useful'

    t1 = ExceptionThread(target=test_function_1, args=[input])
    t1.start()

Bien sûr, vous pouvez toujours gérer l'exception d'une autre façon de la journalisation, comme l'imprimer, ou l'avoir sortie sur la console.

Cela vous permet d'utiliser la classe ExceptionThread exactement comme vous le feriez pour la classe Thread, sans aucune modification spéciale.