Say I want to handle errors by skipping the items with error. I saw https://docs.prefect.io/api/0.5.4/tasks/control_flow.html#filtertask but it doesn’t have an example of how to use it. How do I use FilterTask to be applied for the task results of add so that next task in the flow gets the values that were not skipped?

def skipped(item):
    return isinstance(item, prefect.engine.signals.SKIP) or isinstance(item, prefect.engine.result.NoResultType)

@task
def add(x,y)
   try:
       return x / y
   except:
       raise prefect.engine.signals.SKIP

with Flow("filter-skipped") as flow:
      res = add.map([10, 2, 4], [1, 0, 3])
      # res = FilterTask(res)   #

Hi Jerry - good point, we’ll add an example to the docstring for

FilterTask

. In the meantime, here is how it can be used:

with Flow("filter-skipped") as flow:
      res = add.map([10, 2, 4], [1, 0, 3])
      filtered_res = FilterTask()(task_results=res)

Note that Filtertasks need to first be initialized and then called (which is slightly different from

@task

tasks, which are already initialized); more info on this subtlety can be found here: https://docs.prefect.io/guide/tutorials/task-guide.html