Is it possible to get the flow name through the ta...
# ask-communitya
Adam Everington
11/12/2021, 7:36 AMIs it possible to get the flow name through the task in a state_handler? eg:
Copy code
def my_state_handler(task,new_state,old_state)->None:
send_message(f'the following task failed {task.name} within this flow: {[GET FLOW NAME HERE}]')
@task(on_failed=my_state_handler)
def my_task():
....a
Anna Geller
11/12/2021, 8:43 AM
@Adam Everington yes, you can through Context:
Copy code
prefect.context["flow_name"]🙌 1
Anna Geller
11/12/2021, 9:09 AM
btw, I noticed two more things:
1. it should be on_failure rather than on_failed
2. By using callable rather than state handler, you only need the task and state in the signature, here is an example:
Copy code
from prefect import Flow, task
import prefect
def send_post(task, state):
msg = f'Task {task.name} failed in the flow {prefect.context["flow_name"]}'
print(msg)
@task(on_failure=send_post)
def my_task():
raise Exception("Nope")
with Flow("fail-it") as flow:
my_task()
if __name__ == "__main__":
flow.run()a
Adam Everington
11/12/2021, 9:29 AMTbh, the on_failed was a typo and my state_handler does now represent your example after looking at the documentation more closely 🙂. Wouldn't be a working morning without a question from me right?
a
Anna Geller
11/12/2021, 9:35 AM
nice work! 🙌 no worries, that’s what this channel is for
7 Views