def __init__()

in src/fmeval/transforms/transform_pipeline.py [0:0]

• 30 lines of code
• 6 McCabe index (conditional complexity)

    def __init__(self, nested_transforms: List[NestedTransform]):
        """TransformPipeline initializer.

        :param nested_transforms: A list of Transforms and/or TransformPipelines.
        """
        require(
            isinstance(nested_transforms, List),
            "TransformPipeline initializer accepts a list containing Transforms or TransformPipelines, "
            f"but received an object with type {type(nested_transforms)}.",
        )
        seen_keys = set()
        transform_to_duplicate_keys = defaultdict(list)
        self.transforms: List[Transform] = []
        for nested_transform in nested_transforms:
            if isinstance(nested_transform, Transform):
                for key in nested_transform.output_keys:
                    if key in seen_keys:
                        transform_to_duplicate_keys[nested_transform].append(key)
                    else:
                        seen_keys.add(key)
                self.transforms.append(nested_transform)
            elif isinstance(nested_transform, TransformPipeline):
                self.transforms += nested_transform.transforms
            else:
                raise EvalAlgorithmClientError(
                    f"nested_transform has type {type(nested_transform)}, "
                    "but either Transform or TransformPipeline is expected."
                )
        require(
            len(transform_to_duplicate_keys.keys()) == 0,
            "TransformPipeline contains Transforms with the same output keys as other Transforms. "
            "Here are the problematic Transforms, paired with their offending keys: "
            f"{str(dict(transform_to_duplicate_keys))}",
        )