@ancient-wolf-19325 I want to know how to give to a workflow inputs of Optional arguments and a default argument in same definition if possible?

You can define a Flyte workflow with both Optional arguments and default arguments in the same function signature using standard Python typing and default values. For example, you can use

Optional

from

typing

and assign a default value (e.g.,

None

or another value) to the argument. This is supported for workflow inputs, and the default will be used if the input is not provided at execution time. However, note that for tasks, default values are not automatically recognized unless explicitly provided in the workflow or via a launch plan—this limitation does not apply to workflows themselves, which do support Optional and default values directly in their signature, as shown in the following example:

from typing import Optional
from flytekit import workflow

@workflow
def wf(a: int = 5, b: Optional[str] = None) -> int:
    ...

This pattern is valid and will work as expected for workflows, but not for tasks, due to Flyte's current type system and interface handling. For more, see Non-primitive defaults to workflows and test_default_args_task_optional_int_type_default_none. Would you like more detail or examples on how this works in practice? Sources: - Non-primitive defaults to workflows - test_default_args_task_optional_int_type_default_none 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 Can i give to workflow an input argument of Literal["a", "b", "c"] argument?

No, Flyte does not natively support Python's

typing.Literal

as a distinct type for workflow or task inputs. If you use

Literal<https://github.com/flyteorg/flyte/issues/6410|"a", "b", "c"]

, Flyte will serialize it as a PythonPickle blob, not as a Flyte primitive, and it will appear as such in the UI. The recommended workaround is to use

str

and manually validate allowed values inside your task or workflow. Native support for

Literal

is a known feature request but is not implemented as of now. See [Support

typing.Literal

as flyte type in tasks> and Add support Literal Transformer. Would you like more detail on workarounds or tracking this feature? Sources: - Support `typing.Literal` as flyte type in tasks - Add support Literal Transformer 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 Give me some basic example on how to work around the problem of Literal input argument with a workflow where I validate values inside workflow?

It seems likely that the best workaround is to use a

str

input for your workflow and manually check if the value is one of the allowed options inside the workflow or a task. For example:

from flytekit import workflow, task

@task
def validate_choice(choice: str) -> str:
    allowed = {"a", "b", "c"}
    if choice not in allowed:
        raise ValueError(f"choice must be one of {allowed}, got {choice}")
    return choice

@workflow
def wf(choice: str) -> str:
    return validate_choice(choice=choice)

This approach ensures only valid values are accepted, even though Flyte does not enforce the

Literal

type at the interface level. Would you like more detail or examples for more complex validation? Sources: - https://github.com/flyteorg/flyte/pull/5408 - https://github.com/flyteorg/flyte/issues/590 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 How can i in a workflow take the outputs of tasks and combine them to single list variable that is the output of this workflow. Let's say the tasks return dicts and I want to return list of dicts from the workflow

You can collect the outputs of multiple tasks (each returning a dict) into a list and return that list as the workflow output. Simply call each task, store their outputs in variables, and aggregate them into a list in the workflow body. This is supported and works as expected in Flyte workflows. Example:

from flytekit import task, workflow

@task
def t1() -> dict:
    return {"a": 1}

@task
def t2() -> dict:
    return {"b": 2}

@workflow
def wf() -> list:
    out1 = t1()
    out2 = t2()
    return [out1, out2]

This pattern is valid and will return a list of dicts as the workflow output. Would you like more detail or examples with dynamic numbers of tasks? Sources: - Flytekit test with list outputs 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.