Repository: kedro-org/kedro
File: kedro/pipeline/modular_pipeline.py (Python)

Code before:
        raise ModularPipelineError("Inputs must not be outputs from another node")

    if outputs & free_inputs:
        raise ModularPipelineError("Outputs can't contain free inputs to the pipeline")


def _validate_datasets_exist(
    inputs: AbstractSet[str],
    outputs: AbstractSet[str],
    parameters: AbstractSet[str],
    pipe: Pipeline,
) -> None:
    """Validate that inputs, parameters and outputs map correctly onto the provided nodes."""
    inputs = {_strip_transcoding(k) for k in inputs}
    outputs = {_strip_transcoding(k) for k in outputs}

    existing = {_strip_transcoding(ds) for ds in pipe.datasets()}
    non_existent = (inputs | outputs | parameters) - existing
    if non_existent:
        possible_matches = []
        for non_existent_input in non_existent:
            possible_matches += difflib.get_close_matches(non_existent_input, existing)

        error_msg = f"Failed to map datasets and/or parameters onto the nodes provided: {', '.join(sorted(non_existent))}"
        suggestions = (
            f" - did you mean one of these instead: {', '.join(sorted(possible_matches))}"
            if possible_matches
            else ""
        )
        raise ModularPipelineError(error_msg + suggestions)


def _get_dataset_names_mapping(
    names: str | set[str] | dict[str, str] | None = None,
) -> dict[str, str]:
    """Take a name or a collection of dataset names
    and turn it into a mapping from the old dataset names to the provided ones if necessary.

    Args:
        names: A dataset name or collection of dataset names.
            When str or set[str] is provided, the listed names will stay
            the same as they are named in the provided pipeline.
            When dict[str, str] is provided, current names will be
            mapped to new names in the resultant pipeline.
    Returns:
        A dictionary that maps the old dataset names to the provided ones.
    Examples:
        >>> _get_dataset_names_mapping("dataset_name")
        {"dataset_name": "dataset_name"}  # a str name will stay the same
        >>> _get_dataset_names_mapping(set(["ds_1", "ds_2"]))
        {"ds_1": "ds_1", "ds_2": "ds_2"}  # a set[str] of names will stay the same

Review comment (bug):
Sorting sometimes breaks the order so that the order `non_existent` differs from `possible_matches` found when there is more than one `non_existent` input. For example:
```
def create_pipeline(**kwargs) -> Pipeline:
    pipeline_instance = pipeline(
        [
            node(
                func=split_data,
                # Note wrong input name
                inputs=["model_input_table_NOT_FOUND", "params:model_options"],
                outputs=["X_train_test", "X_test", "y_train", "y_test"],
                name="split_data_node",
            ),
            node(
                func=train_model,
                inputs=["X_train", "y_train"],
                outputs="regressor",
                name="train_model_node",
            ),
            node(
                func=evaluate_model,
                inputs=["regressor", "X_test", "y_test"],
                outputs=None,
                name="evaluate_model_node",
            ),
        ]
    )

    ds_pipeline_1 = pipeline(
        pipe=pipeline_instance,
        inputs=["model_input_table", "X_train_mod"],
        namespace="active_modelling_pipeline",
    )
    ds_pipeline_2 = pipeline(
        pipe=pipeline_instance,
        inputs=["model_input_table", "X_train_mod"],
        namespace="candidate_modelling_pipeline",
    )

    return ds_pipeline_1 + ds_pipeline_2
```

will result this message: `kedro.pipeline.modular_pipeline.ModularPipelineError: Failed to map datasets and/or parameters onto the nodes provided: X_train_mod, model_input_table - did you mean one of these instead: X_train, X_train_test, model_input_table_NOT_FOUND, y_train`

It's not a big problem, but maybe it makes sense to keep grouping order from `get_close_matches`? For example:

```
        for non_existent_input in non_existent:
            possible_matches.append(f"[{', '.join(difflib.get_close_matches(non_existent_input, existing))}]")

        error_msg = f"Failed to map datasets and/or parameters onto the nodes provided: {', '.join(non_existent)}"
        suggestions = (
            f" - did you mean one of these instead: {', '.join(possible_matches)}"
            if possible_matches
            else ""
        )
```

will give a slightly better message: `kedro.pipeline.modular_pipeline.ModularPipelineError: Failed to map datasets and/or parameters onto the nodes provided: X_train_mod, model_input_table - did you mean one of these instead: [X_train, X_train_test, y_train], [model_input_table_NOT_FOUND]`

Wdyt?  🙂

Code after:
        raise ModularPipelineError(
            "Inputs must not be outputs from another node in the same pipeline"
        )

    if outputs & free_inputs:
        raise ModularPipelineError(
            "All outputs must be generated by some node within the pipeline"
        )


def _validate_datasets_exist(
    inputs: AbstractSet[str],
    outputs: AbstractSet[str],
    parameters: AbstractSet[str],
    pipe: Pipeline,
) -> None:
    """Validate that inputs, parameters and outputs map correctly onto the provided nodes."""
    inputs = {_strip_transcoding(k) for k in inputs}
    outputs = {_strip_transcoding(k) for k in outputs}

    existing = {_strip_transcoding(ds) for ds in pipe.datasets()}
    non_existent = (inputs | outputs | parameters) - existing
    if non_existent:
        sorted_non_existent = sorted(non_existent)
        possible_matches = []
        for non_existent_input in sorted_non_existent:
            possible_matches += difflib.get_close_matches(non_existent_input, existing)

        error_msg = f"Failed to map datasets and/or parameters onto the nodes provided: {', '.join(sorted_non_existent)}"
        suggestions = (
            f" - did you mean one of these instead: {', '.join(possible_matches)}"
            if possible_matches
            else ""
        )
        raise ModularPipelineError(error_msg + suggestions)


def _get_dataset_names_mapping(
    names: str | set[str] | dict[str, str] | None = None,
) -> dict[str, str]:
    """Take a name or a collection of dataset names
    and turn it into a mapping from the old dataset names to the provided ones if necessary.

    Args:
        names: A dataset name or collection of dataset names.
            When str or set[str] is provided, the listed names will stay
            the same as they are named in the provided pipeline.
            When dict[str, str] is provided, current names will be
            mapped to new names in the resultant pipeline.
    Returns:
        A dictionary that maps the old dataset names to the provided ones.