Hi, a question about using

with case

and

upstream_tasks

together we have a set of 3 tasks that need to run one after the other (no values shared between them)

A -> B -> C

we also would like to have the ability to run only:

A -> C

that means that

B

should go inside

with case

, but then what happens is that

C

is skipped, because it has

upstream_tasks

related to

B

any solutions available?

Hi Ron, a few options (in increasing complexity): • Set

skip_on_upstream_skip=False

on

C

. Note that this means that

C

will never be skipped, even if

A

is skipped. This may not matter for your use case. • Encode the skipping of

B

inside the

B

task itself.

B

always runs, but some parameter to it skips most of the internals, returning quickly. Simple, and doesn't require any complicated task munging. • Wrap

B

with a

merge

task (with a no-op), so that some "B" always runs. This means that

C

won't be skipped unless

A

is skipped, but is a bit more work. Something like:

from prefect.tasks.control_flow import merge
from prefect import case, task

@task
def noop():
    pass

with case(cond, True):
    b1 = B()
with case(cond, False):
    b2 = noop()

b = merge(b1, b2)

ok, Jim, thanks a lot for the answers, this is really really useful and I’ll choose the channel more wisely in my next question thanks!