https://prefect.io logo
r

Rob Douglas

01/13/2023, 4:03 PM
hey all, I am trying to understand the intended use of `any_successful`: I have a task I want to run when any upstream task succeeds, and to then get the value from it to use. It's not clear to me, though, how to go about finding which
upstream_task
succeeded. Am I on the right track? I've been poking around for a tutorial or example, but am having trouble finding much in the way of examples of
any_successful
Copy code
@task(trigger=any_successful)
def bar(upstream_tasks):
    # get the first upstream_task that succeeded
gratitude thank you in advance for any insights you might have
n

Nate

01/13/2023, 8:15 PM
hey @Rob Douglas
prefect.tasks.control_flow.filter.FilterTask
may be helpful for this
Copy code
from prefect import Flow, task
from prefect.tasks.control_flow.filter import FilterTask

only_successful = FilterTask(lambda x: not isinstance(x, Exception)) # this exact type check may not be appropriate for you

@task   
def add_42(x):
    return x + 42

@task(log_stdout=True)
def log(x):
    print(x)

with Flow("any_successful") as flow:
    parameters = ["lemur", dict, "capybara", 42]
    
    # if you're not mapping, you would construct a list of relevant tasks that may fail
    results = add_42.map(parameters) # results = [TypeError, TypeError, TypeError, 84]
    
    log(only_successful(results)) # prints: [84]
    
if __name__ == "__main__":
    flow.run()
r

Rob Douglas

01/13/2023, 8:54 PM
Thanks for the insights! There's a couple ideas in there I can check out