Hello everyone! I'm new here. I'm handling a new DE role, and after trying the others (Airflow, Luigi, Dagster) I chose prefect for being that weird characteristic of being the most fun to use. It's absolutely amazing. Would anyone know how to capture the exact full python traceback when a task fails? I've set up state_handlers but the State.message only captures an abbreviated form of the error (e.g. ZeroDivisionError) . Thanks!

Hey @Wilhelm Su, happy to hear that! Saw someone do this by creating their own wrapper around

@task

. I think this might work for you.

import traceback
from prefect import task, Flow
from functools import partial, wraps

def custom_task(func=None, **task_init_kwargs):
    if func is None:
        return partial(custom_task, **task_init_kwargs)

    @wraps(func)
    def safe_func(**kwargs):
        try:
            return func(**kwargs)
        except Exception as e:
            print(f"Full Traceback: {traceback.format_exc()}")
            raise RuntimeError(type(e)) from None  # from None is necessary to not log the stacktrace

    safe_func.__name__ = func.__name__
    return task(safe_func, **task_init_kwargs)

@custom_task
def abc(x):
    raise ValueError()
    return x

with Flow("custom-decorator-test") as flow:
    abc(1)

ah of course, an outside wrapper! I'll definitely try this. Thank so much @Kevin Kho!

Actually this might not work. I think you might need to re-raise the error inside the state handler to get the full traceback.

@Wilhelm Su perhaps you could use the repr() of the exception to get you both the exception type and the exception message? For example, this state handler would give the following Slack message:

import prefect
from prefect import task, Flow
from prefect.tasks.notifications import SlackTask
from typing import cast


def post_to_slack_on_failure(task, old_state, new_state):
    if new_state.is_failed():
        if isinstance(new_state.result, Exception):
            value = "```{}```".format(repr(new_state.result))
        else:
            value = cast(str, new_state.message)
        msg = (
            f"The task `{prefect.context.task_name}` failed "
            f"in a flow run {prefect.context.flow_run_id} "
            f"with an exception {value}"
        )
        SlackTask(message=msg).run()
    return new_state


@task(state_handlers=[post_to_slack_on_failure])
def divide_numbers(a, b):
    return 1 / (b - a)


with Flow(name="state-inspection-handler") as flow:
    result = divide_numbers(1, 1)


if __name__ == "__main__":
    flow.run()

Haha, it's because I started trying to use prefect to build stuff even before I checked out all the features. But yeah, I wanted to build a custom state handler that emails me the traceback on fail. That was of course before I saw that email-on-failure is an out of the box functionality of the web ui, and I could just peek at the traceback from the dashboard. Hahaha