Source code for iris.orchestration.validators
from collections import Counter
from typing import Any, List
from pydantic import fields
from iris.io.errors import IRISPipelineError
[docs]
def pipeline_config_duplicate_node_name_check(cls: type, v: List[Any], field: fields.ModelField) -> List[Any]:
"""Check if all pipeline nodes have distinct names.
Args:
cls (type): Class type.
v (List[Any]): Value to check.
field (fields.ModelField): Field descriptor.
Raises:
IRISPipelineError: Raised if pipeline nodes aren't unique.
Returns:
List[Any]: `v` sent for further processing.
"""
node_names = [node.name for node in v]
if len(set(node_names)) != len(node_names):
raise IRISPipelineError(f"Pipeline node name must be unique. Received {dict(Counter(node_names))}")
return v