https://prefect.io logo
#prefect-community
Title
# prefect-community
j

Jerry Thomas

08/23/2019, 10:22 AM
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?
Copy code
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)   #
c

Chris White

08/24/2019, 2:47 AM
Hi Jerry - good point, we’ll add an example to the docstring for
FilterTask
. In the meantime, here is how it can be used:
Copy code
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
💯 1