@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.

@ancient-wolf-19325 Can i run map_task in a task or is it only for workflows?

You cannot run map_task inside a task; map_task is only supported within workflows. Attempting to use map_task inside a task will result in a runtime error, as documented in Flyte issue #3509 and the Flyte map_task documentation. Would you like more detail or an example of correct map_task usage? Sources: - Flyte issue #3509 - Flyte map_task documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 can i return a tuple with a task to workflow

Yes, you can return a tuple from a task to a workflow in Flyte, but the tuple must be strongly typed and its elements must be supported types. For example, you can use typing.Tupleint, str] as a return type. However, returning a tuple that contains a NamedTuple and another value (e.g., (NamedTuple, str)) is not supported and will raise a RestrictedTypeError ([Flyte issue #4358). Would you like more detail or an example of a supported tuple return type? Sources: - Flyte issue #4358 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 Do you have a solution for the following problem with dynamic workflow returns: I am getting a following error:

type_check >> statement_execution_timestamps ~~~~~~~~~~~^^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ File "/app/.venv/lib/python3.12/site-packages/flytekit/core/promise.py", line 480, in __rshift__ if not self.is_ready and other.ref: ^^^^^^^^^ AttributeError: 'dict' object has no attribute 'ref'

With the following code:

@dynamic(pod_template=get_pod_template())
def gemini_financial_statement_extraction(
    limit_pdfs: int = 15000,
    failure_threshold: int = 1000,
    statement_types: list[str] | None = None,
) -> dict:
    """
    Returns:
        dict: Timestamps of the extraction runs (YYYY-MM-DD HH:MM:SS.ffffff format)
    """

    type_check = check_statement_type(statement_types=statement_types)

    statement_execution_timestamps = {}

    for statement_type in statement_types:
        extraction_timestamp = fetch_and_process_documents(
            statement_type=statement_type,
            limit_pdfs=limit_pdfs,
            failure_threshold=failure_threshold,
        )
        statement_execution_timestamps[statement_type] = extraction_timestamp

    type_check >> statement_execution_timestamps

    return statement_execution_timestamps

The error occurs because in a Flyte dynamic workflow, you cannot use the >> operator (which defines dependencies) with a plain Python dict. The right-hand side of >> must be a Flyte Promise or Node, not a dict. In your code, statement_execution_timestamps is a dict of Promises, not a single Promise or Node, so type_check >> statement_execution_timestamps is invalid (Flyte dynamic workflow docs). To enforce dependencies, you should use >> with individual task outputs (Promises) or ensure all dependencies are materialized before returning. For example, loop over the values and use >> for each extraction_timestamp, or collect them in a list and use flytekit's wait_for_input or similar constructs if needed. Returning a dict of Promises is supported, but you cannot use >> with the dict itself. Would you like a step-by-step example of how to enforce dependencies correctly in this context? Sources: - Flyte dynamic workflow docs 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 Yes give example of correct solution with "wait_for_input" and other solutions

To enforce dependencies or pause execution in a Flyte dynamic workflow, you can use wait_for_input for each Promise or ensure all Promises are materialized before returning. Here are two correct approaches: 1. Using wait_for_input for each statement type:

from flytekit import dynamic, wait_for_input

@dynamic
def gemini_financial_statement_extraction(
    limit_pdfs: int = 15000,
    failure_threshold: int = 1000,
    statement_types: list[str] = None,
) -> dict:
    statement_execution_timestamps = {}

    for statement_type in statement_types:
        extraction_timestamp = fetch_and_process_documents(
            statement_type=statement_type,
            limit_pdfs=limit_pdfs,
            failure_threshold=failure_threshold,
        )
        # Wait for external input before proceeding (e.g., human approval)
        wait = wait_for_input(f"wait_{statement_type}", expected_type=bool)
        extraction_timestamp >> wait
        statement_execution_timestamps[statement_type] = extraction_timestamp

    return statement_execution_timestamps

This ensures each extraction waits for an external signal before proceeding (wait_for_input usage). 2. Enforcing dependencies without wait_for_input: If you just want to ensure all tasks complete before returning, collect the Promises and return them directly:

from flytekit import dynamic

@dynamic
def gemini_financial_statement_extraction(
    limit_pdfs: int = 15000,
    failure_threshold: int = 1000,
    statement_types: list[str] = None,
) -> dict:
    statement_execution_timestamps = {}

    for statement_type in statement_types:
        extraction_timestamp = fetch_and_process_documents(
            statement_type=statement_type,
            limit_pdfs=limit_pdfs,
            failure_threshold=failure_threshold,
        )
        statement_execution_timestamps[statement_type] = extraction_timestamp

    return statement_execution_timestamps

Do not use >> with a dict; just return the dict of Promises (dynamic workflow best practices). Would you like more detail on using wait_for_input or other human-in-the-loop constructs? Sources: - wait_for_input usage - dynamic workflow best practices 👍 1 button 👎 0 button 💡 Hint: Mention @ancient-wolf-19325 in the thread for followups.

@ancient-wolf-19325 Does @dynamic workflow support chaining task dependencys with ">>" operator

It seems likely that @dynamic workflows do support chaining task dependencies using the >> operator, but only between individual tasks or subworkflows—not with collections like dicts or lists. The >> operator is used to specify execution order without passing data, and this pattern is valid inside both static and dynamic workflows as long as you use it with Promises or Nodes, not plain Python objects (Flyte chaining entities). If you attempt to use >> with a dict or list of Promises, you will get an error. Always use >> between single task outputs or subworkflow outputs. Sources: - Flyte chaining entities 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 is a task return even though a "dict" also a promise that I can chain

No, a task return value—even if it's a dict—is not itself a Promise; rather, each output from a task in a workflow is a Promise, but the dict object as a whole is not chainable with the >> operator. You can only chain individual Promises (such as a single task output), not a dict or list of Promises (Flyte chaining entities). If you need to chain dependencies, do so with individual outputs, not with the entire dict returned from a task. Sources: - Flyte chaining entities 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.