@ancient-wolf-19325 Can i with conditional in workflow start tasks or only other workflows? Give some examples of the conditional usage

You can use Flyte conditionals in workflows to start both tasks and other workflows. The conditional construct allows you to branch execution based on primitive values, and each branch can invoke a task or a workflow as its action. For example, you can use .then(task()) or .then(workflow()) in the branches of a conditional statement. See the examples below for typical usage patterns. Example of a conditional starting tasks:

from flytekit import task, workflow, conditional

@task
def t1() -> int:
    return 5

@task
def t2() -> int:
    return 10

@workflow
def wf(x: int) -> int:
    return (
        conditional("choose_task")
        .if_(x > 0)
        .then(t1())
        .else_()
        .then(t2())
    )

Example of a conditional starting workflows:

from flytekit import workflow, conditional

@workflow
def wf1() -> int:
    return 1

@workflow
def wf2() -> int:
    return 2

@workflow
def main_wf(x: int) -> int:
    return (
        conditional("choose_workflow")
        .if_(x > 0)
        .then(wf1())
        .else_()
        .then(wf2())
    )

For more details and advanced examples, see the Flyte conditionals documentation and Flytesnacks conditional examples. Would you like more detail or examples, such as using conditionals with task outputs or nested conditionals? Sources: - Flyte conditionals documentation - Flytesnacks conditional examples 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 With the conditional can i run bunch of tasks or only one task at a time?

With Flyte conditionals, each branch (.then, .elif_, .else_) can only execute a single task or workflow—not a sequence of multiple tasks directly. If you want to run multiple tasks as a result of a condition, you should encapsulate them in a sub-workflow or a single task that calls others internally. This is explicitly shown in the Flyte conditionals documentation and Flytesnacks examples. Would you like more detail or an example of how to group multiple tasks under a conditional branch? Sources: - Flyte conditionals documentation - Flytesnacks conditional examples 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 If i have the following workflow that has workflows that return list of dicts and i want to return it combined, how do i do it?

@workflow
def all_transfers(date_tag: str) -> list[dict]:
    """
    Transfers all finance data tables from Snowflake to Postgres.
    Including company mapping table.
    :param date_tag: Date tag for the table to transfer.
    """

    finance_data_transfers_result = finance_data_transfers(date_tag=date_tag)
    company_mapping_transfer_result = company_mapping_transfer(date_tag=date_tag)

    return [*finance_data_transfers_result, *company_mapping_transfer_result]

It seems likely that your approach—using unpacking with *finance_data_transfers_result, *company_mapping_transfer_result]—is correct for combining two lists of dicts in a Flyte workflow, as long as both finance_data_transfers and company_mapping_transfer return list[dict]. Flyte workflows can return lists of dicts, and combining them this way is valid Python and supported by Flyte's type system ([Flyte tasks and workflows type system). If you encounter serialization issues, ensure that the dicts are JSON-serializable and that the return type is properly annotated as list[dict]. Would you like more detail or an example with tasks and subworkflows? Sources: - Flyte tasks and workflows type system 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 I get error with above solution ValueError: o0 is a Promise. Promise objects are not iterable - can't range() over a promise. But you can use [index] or the alpha version of @eager workflows

The error occurs because in a standard Flyte workflow, outputs of tasks or subworkflows (like finance_data_transfers_result) are Promise objects, not materialized Python lists. You cannot iterate or unpack a Promise directly. To combine lists from multiple subworkflows, you must do so inside a task, not in the workflow body (Accessing attributes in workflows). Refactor by creating a task that takes both lists as input and combines them, then call this task from your workflow:

from flytekit import task, workflow

@task
def combine_lists(a: list[dict], b: list[dict]) -> list[dict]:
    return a + b

@workflow
def all_transfers(date_tag: str) -> list[dict]:
    finance_data_transfers_result = finance_data_transfers(date_tag=date_tag)
    company_mapping_transfer_result = company_mapping_transfer(date_tag=date_tag)
    return combine_lists(a=finance_data_transfers_result, b=company_mapping_transfer_result)

Would you like more detail on why this is necessary or how Promise objects work in Flyte? Sources: - Accessing attributes in workflows 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.