Can Luigi throw an exception or return any result?

Question

I use Luigi to run some kind of conveyor. Take a simple example

task = myTask() w = Worker(scheduler=CentralPlannerScheduler(), worker_processes=1) w.add(task) w.run() 

Now let's say that myTask raises an exception at runtime. All I have is a journal from luigi showing an exception.

Is there any way luigi could spread it or at least return failure status?

Then I could make my program respond to this condition.

Thanks.

EDIT I forgot to indicate that the luigi outputs are targeted to the database when I save the result. If an exception occurs, the result is not saved, but the exception does not apply to luigi. I was wondering if luigi has this.

Answer 1

From docs :

Luigi has a built-in event system that allows you to register callbacks for events and run them from your own tasks. You can connect to some predefined events and create your own. Each event descriptor is associated with a Task class and is launched only from this class or subclass. This allows you to easily subscribe to events only from a certain class (for example, for work tasks).

Example:

 import luigi from my_tasks import MyTask @MyTask.event_handler(luigi.Event.FAILURE) def mourn_failure(task, exception): """Will be called directly after a failed execution of `run` on any MyTask subclass """ do_something() luigi.run() 

Luigi has many events that you can choose from . You can also watch these tests to learn how to listen and respond to other events.

Answer 2

What you can do is write the error to a file. For example, in your task, which may fail (call TaskA on it):

 x="" try: do stuff except: x="error!" with open('errorfile.log','w') as f: f.write(x) 

Then, in a task that depends on this error, TaskA will be required for this task. And you can do something like this:

 with open('errorfile.log','r') as f: if f.read()://if anything is in the error log from TaskA //error occurred do stuff else: do other stuff