Weird behaviour in typehinting/type return when using pwf.api.inputs_to*

[ligerzero-ai]

Consider:

from pyiron_workflow import Workflow
import pyiron_workflow as pwf
from pyiron_workflow_lammps.lammps import LammpsInput
from typing import Any, List
wf0 = Workflow("test_mod_dataclass", delete_existing_savefiles=True)
lmp_input = LammpsInput(
    read_data_file="lammps.data",
    pair_coeff=(
        "*", "*",
        "/root/github_dev/test_workflow_nodes/2025_04_29_FeGB_Segregation_Workflows/final_model",
        "Fe"
    ),
    dump_every=10,
    dump_filename="dump.out",
    thermo_every=10,
    min_style="cg",
    etol=0.00001,
    ftol=0.00001,
    maxiter=1_000_000,
    maxeval=1_000_000,
)
# from typing import Any, List
# @pwf.as_function_node("collected_list")
# def collect_to_list(*args: Any) -> List[Any]:
#     """
#     Collects any number of positional arguments into a list.

#     Example:
#         >>> collect_to_list(1, 'foo', True, [3,4])
#         [1, 'foo', True, [3, 4]]
#     """
#     return list(args)

from pyiron_workflow_atomistics.utils import modify_dataclass_multi
@pwf.as_function_node("modded_dataclass")
def modify_dataclass(dataclass_instance, entry_name: str, entry_value: Any):
    from dataclasses import asdict
    from copy import deepcopy
    kwarg_dict = {entry_name: entry_value}
    data = deepcopy(asdict(dataclass_instance))   # deep-copies nested containers
    bad  = set(kwarg_dict) - data.keys()
    if bad:
        raise KeyError(f"Unknown field(s): {sorted(bad)}")

    data.update(**kwarg_dict)
    dataclass_instance = type(dataclass_instance)(**data)
    # re-construct a brand-new instance from the dict
    return dataclass_instance
@pwf.as_function_node("modded_dataclass_multi")
def modify_dataclass_multi(dataclass_instance: Any, entry_names: list[str], entry_values: list[Any]):
    """
    Wraps your single-entry node so you can pass lists of names & values.
    Usage:
      new = modify_dataclass_multi(old, ["a","b"], [1,2])
    """
    if len(entry_names) != len(entry_values):
        raise ValueError("entry_names and entry_values must have the same length")

    ds = dataclass_instance
    for name, val in zip(entry_names, entry_values):
        ds = modify_dataclass.node_function(ds, name, val)
    return ds
wf0.modify_to_rigid_keys_list = pwf.api.inputs_to_list(2, "maxiter", "maxeval")
wf0.modify_to_rigid_values_list = pwf.api.inputs_to_list(2, 0, 0)
wf0.rigid_LammpsInput = modify_dataclass_multi(lmp_input, entry_names= wf0.modify_to_rigid_keys_list, entry_values= wf0.modify_to_rigid_values_list)
wf0.run()

Will fail with:

---------------------------------------------------------------------------
ChannelConnectionError                    Traceback (most recent call last)
Cell In[7], line 61
     59 wf0.modify_to_rigid_keys_list = pwf.api.inputs_to_list(2, "maxiter", "maxeval")
     60 wf0.modify_to_rigid_values_list = pwf.api.inputs_to_list(2, 0, 0)
---> 61 wf0.rigid_LammpsInput = modify_dataclass_multi(lmp_input, entry_names= wf0.modify_to_rigid_keys_list, entry_values= wf0.modify_to_rigid_values_list)
     62 wf0.run()

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/node.py:312, in Node.__init__(self, label, parent, delete_existing_savefiles, autoload, autorun, checkpoint, *args, **kwargs)
    309 # A place for power-users to bypass node-injection
    311 self._setup_node()
--> 312 self._after_node_setup(
    313     *args,
    314     delete_existing_savefiles=delete_existing_savefiles,
    315     autoload=autoload,
    316     autorun=autorun,
    317     **kwargs,
    318 )

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/node.py:357, in Node._after_node_setup(self, delete_existing_savefiles, autoload, autorun, *args, **kwargs)
    354             self.load(backend=autoload)
    355             break
--> 357 self.set_input_values(*args, **kwargs)
    359 if autorun:
    360     with contextlib.suppress(ReadinessError):

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/io.py:447, in HasIO.set_input_values(self, *args, **kwargs)
    444 self._ensure_all_input_keys_present(kwargs.keys(), self.inputs.labels)
    446 for k, v in kwargs.items():
--> 447     self.inputs[k] = v

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/io.py:118, in IO.__setitem__(self, key, value)
    117 def __setitem__(self, key: str, value: Any) -> None:
--> 118     self.__setattr__(key, value)

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/io.py:94, in IO.__setattr__(self, key, value)
     92 def __setattr__(self, key: str, value: Any) -> None:
     93     if key in self.channel_dict:
---> 94         self._assign_value_to_existing_channel(self.channel_dict[key], value)
     95     elif isinstance(value, self._channel_class):
     96         if key != value.label:

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/io.py:110, in IO._assign_value_to_existing_channel(self, channel, value)
    108 def _assign_value_to_existing_channel(self, channel: OwnedType, value: Any) -> None:
    109     if isinstance(value, HasChannel):
--> 110         channel.connect(value.channel)
    111     else:
    112         self._assign_a_non_channel_value(channel, value)

File ~/miniconda3/envs/pyiron_workflow/lib/python3.12/site-packages/pyiron_workflow/channels.py:147, in Channel.connect(self, *others)
    145         other.connections.insert(0, self)
    146     else:
--> 147         raise ChannelConnectionError(
    148             self._connection_conjugate_failure_message(other)
    149         ) from None
    150 else:
    151     raise TypeError(
    152         f"Can only connect to {self.connection_conjugate()} "
    153         f"objects, but {self.full_label} ({self.__class__}) "
    154         f"got {other} ({type(other)})"
    155     )

ChannelConnectionError: The channel /test_mod_dataclass/modify_to_rigid_keys_list.list (<class 'pyiron_workflow.mixin.injection.OutputDataWithInjection'>) has the correct type (<class 'pyiron_workflow.channels.OutputData'>) to connect with /modify_dataclass_multi.entry_names (<class 'pyiron_workflow.channels.InputData'>), but is not a valid connection.Please check type hints, etc. /test_mod_dataclass/modify_to_rigid_keys_list.list.type_hint = <class 'list'>; /modify_dataclass_multi.entry_names.type_hint = list[str]

Whereas removal of the typehints from:

def modify_dataclass_multi(dataclass_instance: Any, entry_names: List, entry_values: List):

into

def modify_dataclass_multi(dataclass_instance: Any, entry_names, entry_values):

Will allow successful execution

[ligerzero-ai]

I am guessing that there is some kind of weird type conversion in the inbuilt functions that messes with the typehinting, as it says the two types of list and List returned internally by pwf are different somehow.

[liamhuber]

Yeah, the problem is that pwf.api.inputs_to_list is generating a node whose output is typed but with insufficient specificity. modify_dataclass_multi is expecting something of type list[str], but wf0.modify_to_rigid_keys_list is merely providing list. The data it holds is perfectly compatible, because you've initialized it all to strings, but we never get to discover that because we are never even allowed to form the connection.

Explicit solution

One thing I see here that I definitely want to do is provide a new kwargs on the metanode creator for specifying the content type, i.e.:

import pyiron_workflow as pwf

n_untyped = pwf.api.inputs_to_list(2, "maxiter", "maxeval")
print(n_untyped.outputs.list.type_hint)
>>> list

n_typed = pwf.api.inputs_to_list(2, "maxiter", "maxeval", content_type_hint=str)
print(n_typed.outputs.list.type_hint)
>>> list[str]

This is an easy fix. I expect I can get it patched today and am hoping to make a new release (with #678) this week. This is a useful idea for the dict tranformer too, so I'm very grateful you caught this.

Implicit solution

Another idea I see is to infer a content type hint at call-time based on the values of the arguments. Something like

import pyiron_workflow as pwf

print(
    pwf.api.inputs_to_list(
        2, "maxiter", "maxeval", infer_type=True
    ).outputs.list.type_hint
)
>>> list[str]

print(
    pwf.api.inputs_to_list(
        2, "maxiter", 42.0, infer_type=True
    ).outputs.list.type_hint
)
>>> list[str | float]

I'm more ambivalent about this. It could be extremely convenient, especially if it defaulted to True, but also "explicit is better than implicit". What do you think, @ligerzero-ai?

Something like if content_type_hint is not NotData use it; elif infer_type infer it; else no sub-hint

Temporary workaround

At any rate, in the meantime you can resolve it at application time by overriding the type hint after creation:

import pyiron_workflow as pwf

@pwf.as_function_node
def need_string_list(x: list[str]):
    return x

wf = pwf.Workflow("runtime_runaround")
wf.list_producer = pwf.api.inputs_to_list(2, "maxiter", "maxeval")
wf.list_producer.outputs.list.type_hint = list[str]  # Override
wf.list_consumer = need_string_list(wf.list_producer)

[liamhuber]

Note to self: generating the hints programatically is straightforward, nicely handles duplicates, and plays nicely with our type hinting:

from typing import Union

subset = int | str
superset = int | str | float
types = (str, int, str, str, int)
hint = Union[*types]
print(hint)
>>> typing.Union[str, int]

from pyiron_workflow.type_hinting import type_hint_is_as_or_more_specific_than
print(
    type_hint_is_as_or_more_specific_than(subset, hint),
    type_hint_is_as_or_more_specific_than(superset, hint),
)
>>> True False

[ligerzero-ai]

I have no idea - personally I think having the option for inferring the typehints would be useful. The workflow language is already considerably a steeper learning curve than most people will tolerate if they aren't really into coding (I think).

I will say that I have lost more time on typehints than I would have liked that aren't strictly attributed to good coding practice - and I consider my patience to be fairly high, compared to a PhD student or something like that.

Seems like a policy decision to me. I would cast a vote for having it by default infer and let power users turn it off. If you want to have the infer off by default, a nicer error would help a lot - like TypeHint might not be specific enough - specify the datatype in flag ...

[liamhuber]

@ligerzero-ai, I won't get to this today but am still hoping for tomorrow. I'm happy to wire up automatic inference as the default.

[ligerzero-ai]

no problem mate - it's really not urgent 👍