Wilhelm Su

    Wilhelm Su

    10 months ago
    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!
    Kevin Kho

    Kevin Kho

    10 months ago
    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)
    Wilhelm Su

    Wilhelm Su

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

    Kevin Kho

    10 months ago
    Actually this might not work. I think you might need to re-raise the error inside the state handler to get the full traceback.
    What are you trying to do with the state handler? Log the whole error to send somewhere?
    Anna Geller

    Anna Geller

    10 months ago
    @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()
    @Wilhelm Su and if you want to get a full exception traceback, you can use something like this: https://gist.github.com/2014180ee5eaec9ea54f4d3f5b98ca93 This should give you:
    Wilhelm Su

    Wilhelm Su

    10 months ago
    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