https://prefect.io
Join Slack
Is there a way to make a state handler do something different if a specific task fails? For example...
# ask-community
k
Is there a way to make a state handler do something different if a specific task fails? For example, I have a chain of tasks that depend on each other: 
a -> b ->c
I want it so that if 
a
and 
b
fails, it sends a Slack notifications saying “Flow failed”. But if 
c
fails, I want it alter the Slack message by adding b’s output
k
I think what you want is for C to just have it’s own state handler?
Or this is much easier expressed as a SlackTask with 
b
as an input and 
c
as an upstream with trigger 
any_failed
k
I agree, I think the latter option sounds easiest. But if I wanted to have a separate state handler for a specific task, how would I be able to do that?
k
You…write a new function? Not seeing why it would be different than the first? You specify it anyway at the task level right?
Copy code
@task(state_handers=[...])
k
Oh! This whole time I’ve only been setting state handler at the Flow level
Thanks for the tip!
k
Oh lol!
k
Is there a way I can pass on an output of a task into the state handler?
Like if a task fail, I want the state handler to include the value of a variable within the task
k
Let me draft an example
❤️ 1
Did you get your swag btw?
k
Still waiting on it 😭
k
What!? Let me check on that too
It’s not pretty though because you basically need to embed it in the result
Copy code
from prefect import Flow, task
import prefect

class SomeError(ValueError):
    def __init__(self, *args, x):
        super().__init__(*args)
        self.x = x

def st(task, old_state, new_state):
    if new_state.is_failed():
        <http://prefect.context.logger.info|prefect.context.logger.info>(type(new_state.result))
        <http://prefect.context.logger.info|prefect.context.logger.info>(new_state.result.x)
    return new_state

@task(state_handlers=[st])
def abc(x):
    # we want to keep x in state handler
    if x == 2:
        raise SomeError("test error", x=x)
    return x+1

with Flow("..") as flow:
    abc(2)

flow.run()
Ah ok I see order is processing on the swag
k
as long as its not lost in the mail, I am happy 😃
k
You don’t have to do these code shenanigans if you use a downstream task
k
Yeah youre right
I have a task that outputs a string, I only want it to trigger a downstream task if that string has length > 0. If the downstream task is triggered, I want it to print the string from the upstream task I currently have it as
Copy code
if len(str) > 0: 
  raise FAIL 
return str
But it seems like when I raise FAIL, the task returns 
None
instead
How do I return an output as well as raise FAIL signal to trigger the downstream task? Or better yet, is there a simpler way I could be doing this?
k
SKIP
propagates to downstream (and FAIL does too so the downstream shouldn’t run)
k
Oh I forgot to mention my downstream task has trigger=any_failed
So you’re suggesting I should use 
raise SKIP
instead?
k
That might be an option for this use case
k
Would a SKIP trigger a downstream flow if the downstream flow has trigger=any_failed?
I also am still unclear on whether using SKIP instead of FAIL would allow the function to return a value
Oh actually I think I might be misunderstanding you, lemme try some stuff out
k
Ah SKIP is successful sorry I had it mixed up in my head. I think you are trying to do B -> C -> any_failed_task where some value is passed from B to C to the any_failed_task. I am suggesting to make B an input into the any_failed task
k
Yep that makes sense, thanks!
5 Views
Open in Slack
PreviousNext
Powered by

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.

Powered by