Hi All ... prefect newbie here. Sorry if the answer to my question is obvious and somewhere. But I couldn't find it. I am struggling to figure out how to access the state of a task (task a) within another task (task b). I want task b to always run whether or not task a succeeds or fails. I want task b to do-stuff if task A fails, or skip if task A succeeds.  Sort of like this

@task 
def a ():
   < do something that might succeed or fail >
@task (trigger = always_run)
def b():
  < if task a status==FAIL .. do something >
  < if task b status=SUCCESS .. skip >    

with Flow as flow:
    res1 = a()
    b(res1)
flow.run()

This is part of a larger flow with other tasks ahead and after a&b. I tried

trigger=any_failed

on task b, but that causes task b to fail if task a succeeds (due to trigger not satisfied) .. which is not what I want. I want task b to always show success. Again .. sorry if this is obvious and its just a newbie-thing. Feel free to just point me to the right place in the doc. Thanks in advance

Hi @goodsonr, welcome to the Prefect club! There are a couple of different ways to achieve what you're aiming for: • if what you want to do is just take an action to handle Task A's transition to a failed state, something like a state handler would do the trick here. • if you want to treat Task B as its own unit of work, things become a bit more complex. You'd likely need to create a custom trigger where if the upstream succeeds or skips, you raise a skip signal. If the upstream fails, go ahead.

more interested in option 2. Was hoping there was some way to detect the state of task A within Task B and raise the signal.SKIP within B .. as opposed to a developing a new/custom trigger. If not, oh well... Thanks for the response

Unfortunately, I don't think there's a plug-and-play solution using just Prefect Core at the moment. If you want to use an orchestration layer, you could make a GraphQL call in Task B to achieve the same results. Otherwise a custom trigger will be your best bet.

@goodsonr I think @Zachary Hughes’s suggestion of either querying the API for the upstream state or using a trigger to handle logic is probably the “correct” way in Prefect. Tasks use triggers to decide how to react to upstream states, and then use their logic to work with the results of those upstream states. I think your use case — take an action if upstream fails and skip otherwise — is actually common enough that we may introduce built-in triggers to handle it. However, here’s what a custom trigger would look like, if it’s helpful:

wow... thanks very much for that. most helpful in my understanding ... which is always a good start! I'll see if I can make-it-work. tks again

:yes: I may have slightly over-complicated things versus your actual use case but hopefully this gives you a jumping-off point for epxloring!

@Jeremiah - thanks again. This works .. almost. I'm finding that

list(upstream_states.values())

is not always coming back in the same order over multiple runs, so that the most recent task is not always first in the list. Perhaps I'm still missing something... or are these values coming from a

dict

edit -> adding to my own comment .. in my test script I am passing more than just

a

to

def b(a)

. I am also passing a couple of parameters .. and i can see these are causing the ordering-grief. I can work around that so all is AOK for now.

Got it - yes my example only shows this working with one upstream task. However,

upstream_states

is a dictionary keyed by

{Edge: State}

, so if you look at key edge’s

upstream_task

you can see which task you’re working with. For example, in my code, I could have asserted

list(upstream_states)[0].upstream_task is a_result

- in other words, the first key of the

upstream_states

dict is an edge whose upstream_task was

a_result

. You could also create logic based on task name, slug, or any other attribute