How to create a conditional flow?

[newskooler]

I have the following flow below, but I cannot get the if / else (check) condition to work.
I read this https://docs.prefect.io/core/idioms/conditional.html but it did not help making my code work unfortunately.
Can someone please help me get to the bottom of this?

import prefect
from perfect import task, case, Flow, Parameter
from prefect.engine.executors import LocalDaskExecutor
from prefect.environments import LocalEnvironment

@task
def check_task(input) -> bool:
    if input == 2:
        raise prefect.engine.signals.SKIP()
    else:
        return (1, 2, 3)

@ task 
def main_task(input_tuple) -> tuple:
    # do some task


with Flow("example") as flow:
    input = Parameter("input", required=True)
    check_result = check_task(input)
    with case(check_result, tuple):
        raw_data_info = main_task(check_result)

    flow.environment = LocalEnvironment(
        labels=[], executor=LocalDaskExecutor(scheduler="threads", num_workers=num_workers),
    )

[jcrist]

Hi @snenkov

When posting issues, please:

• Ensure that your example can be run, there's some clear typos in the above (thomasfrederikhoeck isn't a valid decorator :))
• Describe what the expected behavior is, from the above it's not clear to me what you want this flow to do
• Describe how it doesn't work. From the above, it's not clear how things fail.

Doing these things makes it much easier for people to help you.

---

From a quick glance over, it looks like you expect case to match on the type of the value (tuple). case matches on equality - you probably want something like:

import prefect
from perfect import task, case, Flow, Parameter

@task
def check_task(input) -> bool:
    if input == 2:
        raise prefect.engine.signals.SKIP()
    else:
        return (1, 2, 3)

@task
def main_task(input_tuple) -> tuple:
    # do some task

@task
def is_tuple(x):
    return isinstance(x, tuple)


with Flow("example") as flow:
    input = Parameter("input", required=True)
    check_result = check_task(input)
    with case(is_tuple(check_result), True):
        raw_data_info = main_task(check_result)

[newskooler]

I will do my best to be more accurate next time : ) Sorry for the mess.
Yeah, this seems like exactly what I was looking for. Thank you @jcrist