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.

Prefect Community

Hi, I'd like to write a task that is triggered on triggers.always_run, but take two different actions depending on if any tasks failed, or if all tasks were successful. Any pointers?
As in something similar to:
```@task(trigger=prefect.triggers.always_run)  # one new line
def tear_down_cluster(cluster):
    if all_successful:
        print(":)")
    else:
        print(":-/")```

Hey <@U013XTS1ZT8>,
I think the best way to approach this is to add some downstream dependent tasks that take actions based on the final state of upstream tasks, rather than attach an `always_run` trigger to a task with conditional logic. <https://docs.prefect.io/api/latest/triggers.html#functions|Here’s a great example> in the documentation of taking two courses of action based on the final state of `task_a`, giving you granular visibility across the final states of your tasks and the logic engaged.

You’ll notice you can set the dependency with the `upstream_tasks` kwarg to ensure that task always runs first.

Hey Kyle - ok, I noticed that. The reason I was looking at the always_run was I wasn't 100% sure how to guarantee that a task is going to run, regardless of other failures, and always_run seemed to be the way to do this.
What would a way to do this be setting one task to trigger on any_failed, and another to trigger on all_succesdful. Naturally I don't want to ever enter a case where both of these tasks are fired, it needs to only ever be one.

As in, I don't really care what task has failed, if any of them did, then I want a notification. Otherwise I want a (different) notification, but only ever a single notification.

Hmm, in that case you may be better off with a<https://docs.prefect.io/core/concepts/notifications.html#sending-a-simple-notification| Flow Level State Handler >that fires based on the final state of a flow. If you configure your tasks to bubble their failure to the Flow Level, then you can send one of those two notifications based on the final state of the flow. You can also set a State Handler notification at the task level as needed, that looks to the final state of a task(s) instead.

Something like:
```def post_to_slack(task, old_state, new_state):
    if new_state.is_successful():
        pass
    if new_state.is_failed():
        pass
    return new_state```

Haha, so this is what I started out with believe it or not. But there was a problem with all my states being None. I'll try it again... I might have messed something up.

ok- this seems to work, I must have introduced a bug somehow previously. I was getting None on final states similar to <https://github.com/PrefectHQ/prefect/issues/1307> - this is what led me to the triggers solution. Thanks Kyle!