Created
November 21, 2021 21:59
-
-
Save anna-geller/2014180ee5eaec9ea54f4d3f5b98ca93 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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