Below is an example of how to wrap the task function to get the full traceback when a task errors. The custom_task
can also be used to set default values for max_retries
or state_handlers
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)