Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Why is it when a task is set to only trigger if it's upstream fails the said task is then marked as failed? The task was never executed, shouldn't it be marked as skipped?

I suppose the same is try for all_successful trigger, as in if the upstream failed the downstream tasks are marked as failure, but again they didn't actually run

so would have thought they would be marked as skipped

I suspect this is because `reference_tasks`, which determines the state of the flow as a whole, would then be `Skipped`? when in principle it's failed.

I believe if you have 1 task as failed and the others skipped/successful the whole flow would be marked as failed. If not you could just add it to the reference tasks array?

`flow.set_reference_tasks([failedTask, skippedTask, skippedTask])`

I dunno, just seems odd to me that a task is marked as failure when it didn't even execute

<@U018S5Y8EUX> Matt is correct - this is to ensure the final flow state reflects the presence of failure; if you want the behavior you describe, simply use a different trigger on your tasks so that Failure of an upstream task is converted to a `Skipped` state on the downstream tasks

Sorry, not sure I follow, how would you convert a state to "Skipped"?

You can use a signal within the trigger like so:
```from prefect import signals

def my_skipped_trigger(upstream_states: Dict[Edge, State]):
    if any(s.is_failed() for s in upstream_states.values()):
        raise signals.SKIP("Some upstreams failed")

@task(trigger=my_skipped_trigger)
…```

<@ULVA73B9P> archive “How to make my tasks skip instead of triggerfail anytime an upstream task fails”

<https://github.com/PrefectHQ/prefect/issues/3234>

I actually get an error when trying to run this : `def my_skipped_trigger(upstream_states: Dict[Edge, State]):`
`TypeError: 'SignatureValidator' object is not subscriptable`

Here is my function trying to use trigger:

`@task(name = "Unhappy Path", result=LocalResult(dir='~/Desktop/Credit/results'), trigger = my_skipped_trigger)`
`def netting_data_failed():`
    `logger = prefect.context.get("logger")`
    `<http://logger.info|logger.info>("Caught failed flow!")`

`def my_skipped_trigger(upstream_states: Dict[Edge, State]):`
    `if any(s.is_failed() for s in upstream_states.values()):`
        `raise signals.SKIP("Some upstreams failed")`

`from prefect.tasks.core.collections import Dict`

There are other tasks etc in my code but this is just what I've added

OH, I see - that `Dict` should be imported from `typing`: `from typing import Dict`; the type annotation is not necessary, I just provided it so you knew what the argument task type was