How to handle a List of "Input" objects in a script function ?

[sam2x]

What is the best way to pass list of "Input" arguments (some fields of my Input class are Artifact) ?
I have the following case:

• An "Input" inherited class "MyInput"
• Two tasks "A" and "B" which return an "Output" inherited class "MyOutput"
• A third task "C" which aims to "aggregate/merge" a list of "Input"

I can't play with kargs (due to inherent hera script arguments logics) or multiple arguments (would raise error since i can pass only one argument with Script-runner feature):

Example when I try this:

        merge = mylib.ctn_helper_merge(name="merge", arguments={
            "inputs":[
                { "some_struct": {bla},
                  "data": taskA.get_artifact("data").with_name("data")
                },
                { "some_struct": {bla},
                  "data": taskB.get_artifact("data").with_name("data")
                },
                {...},]}
[...]
def ctn_helper_merge(inputs: List[MyInput]):
...

When I try to pass this list inside arguments of the task, I have correctly the "some_struct" accessible, however "data" Artifact Object contains the "full" attribute Hera Artifact class instead of the loaded artifact, so it s not "transposed/interpreted" correctly by Hera stub when compiled.

I was expecting it to link and load the TaskX artifact.

Any pointers please ?

Thank you,
Sam

[sam2x]

I have dig a bit into this. Apparently Hera doesn't support "nested" object (or more precisely List of Input). That would be a game changer to me and all if Hera would let me build dynamically "Parameter/Artifact". In my case, I don't know in advance how many "Input" are provided. Using "List" of Input approach bring a lot of advantage/use-case (intersection/processing of data result coming from an unknown number of task, but also inscreasing performance).

The relevant code is located in "/src/hera/workflows/script.py", in the function _get_inputs_from_callable:
And the logic is mentioned function's description:

"""Return all inputs from the function.

This includes all basic Python function parameters, and all parameters with a Parameter or Artifact annotation.
For the Pydantic IO experimental feature, any input parameter which is a subclass of Input, the fields of the
class will be used as inputs, rather than the class itself.

Note, the given Parameter/Artifact names in annotations of different inputs could clash, which will raise a ValueError.

The key point (that was already mentioned by @elliotgunton on my last issue) make sense. "Input" itself is not mapped, but the attributes having Input subclass are. I understand why my attempts fail.

Would you consider adapting the logic to the following approach?

1. Detect if annotated, retrieve the type
2. Detect if current type is "List" and if element type is specified
3. If yes, detect if it's type "InputV1/InputV2"
4. For any InputV1/InpuV2, iterate over the class fields add map directly them as root inputs

I believe no change would be required later on, since Parameter/Artifacts have been standardized in specific inputs arrays, so the routine should detect any missing ones from user side.

So, this would cover the following case:

class MyInput(Input):
    static_int: int
    my_artifact: [Artifact(name='my-artifact', path="/a/loaded/path")] 

@script(runner)
def test(args: List[MyInput]):
     for arg in args:
          # I can operate logic there      
          pass
# and also annotated one...
@script(runner)
def test(args: [List[MyInput], "test"]):
     for arg in args:
          # I can operate logic there      
          pass

What do you think ? I may help on this.

Br,
Sam

[sam2x]

Hey,

Providing an update after analyzing in details the code. Currently, not only the previous function needs to be modified, but either an array-capabilities would be used or a naming convention would be defined to avoid conflict since all parameters are mapped into the callable function. So, this is more tricky than expected and require more important change, i discontinued this approach.

Therefore, and until more features gets included along Hera versions, I use an alternative solution: using a temporaring shared storage for each tasks(item of my list), and then passing a list of related ID pointing to those filename for my merge. More safe and conflict-proof, since I can call multiple same tasks dynamically and just pass the generated id to my aggregating task.

Br,
Sam

[alicederyn]

Can you give a bit more information about your exact use-case? For instance, do you know at the time you're creating your workflow how many parameters/artifacts you have, and you just want a script function that can take a varying number of inputs for reusability/readability? Or do you only discover the number of artifacts as part of running the workflow? Perhaps you could give a complete skeleton workflow that roughly matches your goal?

[sam2x]

Hello,

I don't know how many parameters/artitifacts, else I would define a specific Input class. Basically I wanted define a list of a specific class inherited from Input, that could be feeded by one specific task without hardcoding each of them. All the interest was to pass that list (of unknow length) and process it. And I can't use loop, unless I keep a reference of previous operation. Does it make sense ?

Thank you
Sam

[sam2x]

I have a "transpiler/bridge" that use Hera routine and build/call dynamically data*

[alicederyn]

So you're also outputting a list of artifacts from a single previous step?

[sam2x]

Outputing from multiple tasks one artifact that is declared in an Output class. The class looks like:

class TaskOutput(Output):
    data: Annotated[Path, Artifact(name="data_user", archive=NoneArchiveStrategy(), path="/tmp/data.json")] = "/tmp/data.json"
    [...]

Multiple parallel tasks generate data and pass it all in an aggregator task (the one i want to handle List) .
DAG example looks like:

[taskA,taskB,taskX] > aggregator

I was assuming the following definition would just work

def aggregator(inputs: List[MyInput])

I believe that would be a very useful use-case to support this. Benefits i see so far:

• User friendly data processing from Hera (i saw it was possible to construct such approach via yaml directly)
• Performance (dealing with list instead of each input using one task ressource only)
• Don't have to use "temporary storage" alternative
• Don't have to implement data isolation per "workflow" on my own.

Challenge from hera perspective:

• Avoid naming conflict since "Input" are mapped in the function definition
• Implement a List-Input capable parameter (maybe by introducing new runner type such as ListInput/ListOutput ? )
• Support Annotation of this List/nested parsing ?