How to catch specific errors when running a task?

[marcietran]

I would like to catch an AssertionError that is raised within my task, but it seems I can only catch generic RuntimeErrors, and I can't figure out how to see what other errors may have been raised before it (programmatically)

For example, this will not work:

import d6tflow

class TaskExample(d6tflow.tasks.TaskCache):
    def run(self):
        result = 5 + 5
        assert result > 100, "result must be greater than 100"
        self.save(result)

try:
    d6tflow.run(TaskExample())
except AssertionError:
    # do some special logic here for this particular error
    print("Exception caught")

This will work but it's not desirable to catch all generic RuntimeErrors

import d6tflow

class TaskExample(d6tflow.tasks.TaskCache):
    def run(self):
        result = 5 + 5
        assert result > 100, "result must be greater than 100"
        self.save(result)

try:
    d6tflow.run(TaskExample())
except RuntimeError:
    print("Exception caught")

I tried using the traceback library to get the stack trace but I could only get the stack trace of the RuntimeError, which is not very useful. Is there any way to catch a specific error or at least get the last error that was raised before RuntimeError?

Edit: Also, when you except a RuntimeError, the stack trace still prints and I'm wondering if there's a way to suppress that

[d6tdev]

Thanks for raising the issue, we'll need to take a look. You can hide execution output but if there is an error it will still show.

[d6tdev]

@marcietran since the tasks gets executed in workers it wasn't as straight forward to reraise the original error, we're still looking into it.