Hi All ... prefect newbie here. Sorry if the answe...
# prefect-community
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
Copy code
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()
This is part of a larger flow with other tasks ahead and after a&b. I tried
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:
Copy code
import prefect
from prefect import task, Flow, Parameter

def run_on_failed_or_skip_trigger(upstream_states):
    a_state = list(upstream_states.values())[0]
    if a_state.is_failed():
        return True
        raise prefect.engine.signals.SKIP()

def a(outcome):
    if outcome == 'succeed':
        return True
        raise ValueError(outcome)
def b(a):
    print("Running B!")

with Flow("example") as flow:
    outcome = Parameter('outcome')
    a_result = a(outcome)
    b_result = b(a_result)
The custom trigger pulls `a`’s state out of the upstream states dictionary and proceeds (returns True) if it’s failed, or raises a SKIP signal otherwise (which I think is your desired outcome, modify if not!)
The reason I wanted to pull this code though is that it illustrates something (advanced) about Prefect — even failed states have
that are passed to downstream tasks. By default, I think it’s the contents of the error, but it can be arbitrary if you use a Prefect signal. So if you examine the
statement inside
, you’ll see that we can examine the failed “result” of
. This is another way that you could build your logic, though I confess it didn’t occur to me until I was putting this together — B can have an
trigger and examine the returned value of A, and take action appropriately
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
👌 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
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
edit -> adding to my own comment .. in my test script I am passing more than just
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,
is a dictionary keyed by
{Edge: State}
, so if you look at key edge’s
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
dict is an edge whose upstream_task was
. You could also create logic based on task name, slug, or any other attribute