Skip to content

Instantly share code, notes, and snippets.

@anna-geller
Created November 21, 2021 21:59
Show Gist options
  • Save anna-geller/2014180ee5eaec9ea54f4d3f5b98ca93 to your computer and use it in GitHub Desktop.
Save anna-geller/2014180ee5eaec9ea54f4d3f5b98ca93 to your computer and use it in GitHub Desktop.
import prefect
from prefect import task, Flow
from prefect.tasks.notifications import SlackTask
from typing import cast
import traceback
def exception_to_string(excp):
# from https://stackoverflow.com/questions/4564559/get-exception-description-and-stack-trace-which-caused-an-exception-all-as-a-st/58764987#58764987
stack = traceback.extract_stack()[:-3] + traceback.extract_tb(
excp.__traceback__
) # add limit=??
pretty = traceback.format_list(stack)
return "".join(pretty) + "\n {} {}".format(excp.__class__, excp)
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} \n"
f"and a full exception traceback: {exception_to_string(new_state.result)}"
)
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment