Is it possible to get the flow name through the task in a state_handler? eg: ```def my_state_handler...

Adam Everington

11/12/2021, 7:36 AM

Is 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():
  ....

Had a look through the task class on github and I can see it's passed to various methods but couldn't see a property it persisted in

Anna Geller

11/12/2021, 8:43 AM

@Adam Everington yes, you can through Context:

Copy code

prefect.context["flow_name"]

More on that here: https://docs.prefect.io/api/latest/utilities/context.html#context-2

🙌 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()

Adam Everington

11/12/2021, 9:29 AM

Tbh, 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?

Anna Geller

11/12/2021, 9:35 AM

nice work! 🙌 no worries, that’s what this channel is for

6 Views

Open in Slack

Previous Next

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.