Is there a way to get the logs to show the line in my code that caused an issue? In several cases I get Prefect-specific errors and the stack trace is through the Prefect file chain, but doesn't indicate which line in my code I could tweak/remove to avoid an issue. There a way to do this? Also, the logs to the screen seem to exclude more complex outputs at times, presumably due to formatting, where's the default save location of them for review after the fact without these constraints?

@Tom Shaffner you could use a state handler to send you an email or a Slack message upon failure. You could then use a similar logic to this to print the full exception traceback. I didn’t understand the second question - are you asking what is a default location of logs?

My company doesn't have slack, and it restricts email access very strictly due to past phishing issues. I'll look into your example though, thanks. And yes, in the second case.

Maybe you can then just log the full exception traceback?

def log_exception(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)
        logger = prefect.context.get("logger")
        logger.error(exception_to_string(new_state.result))
    return new_state


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

The logs are stored in the backend and you could retrieve those using a GraphQL query:

query {
	flow_run{
    logs {
      message
      level
      info
      created
    }
  }
}

I saw someone make their own task decorator like this

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)

Woah @Kevin Kho, that's cool! Adding that now...