Repository: kedro-org/kedro
File: kedro/framework/cli/utils.py (Python)

Code before:


def _split_params(ctx, param, value):
    if isinstance(value, dict):
        return value

    if not value:
        return {}
    param_list = value.split(",")

    # The input is expected to be in OmegaConf dotlist format
    conf = OmegaConf.from_dotlist(param_list)

    # Convert the OmegaConf object to a plain dictionary
    return OmegaConf.to_container(conf, resolve=True)


def _split_load_versions(ctx, param, value):
    lv_tuple = _get_values_as_tuple([value])
    return _reformat_load_versions(ctx, param, lv_tuple) if value else {}


def _get_values_as_tuple(values: Iterable[str]) -> tuple[str, ...]:
    return tuple(chain.from_iterable(value.split(",") for value in values))


def _deprecate_options(ctx, param, value):
    deprecated_flag = {
        "node_names": "--node",
        "tag": "--tag",
        "load_version": "--load-version",
    }
    new_flag = {
        "node_names": "--nodes",
        "tag": "--tags",
        "load_version": "--load-versions",
    }
    shorthand_flag = {
        "node_names": "-n",
        "tag": "-t",
        "load_version": "-lv",
    }
    if value:
        deprecation_message = (
            f"DeprecationWarning: 'kedro run' flag '{deprecated_flag[param.name]}' is deprecated "
            "and will not be available from Kedro 0.19.0. "
            f"Use the flag '{new_flag[param.name]}' instead. Shorthand "
            f"'{shorthand_flag[param.name]}' will be updated to use "
            f"'{new_flag[param.name]}' in Kedro 0.19.0."
        )
        click.secho(deprecation_message, fg="red")

Review comment (suggestion):
I'm not sure `omegaconf` does equally strict error handling and I would expect us to still provide the error messages above when a user doesn't provide a proper parameter key value pair. 

Code after:
def _split_params(ctx, param, value):
    if isinstance(value, dict):
        return value
    dot_list = []
    for item in split_string(ctx, param, value):
        equals_idx = item.find("=")
        if equals_idx == -1:
            # If an equals sign is not found, fail with an error message.
            ctx.fail(
                f"Invalid format of `{param.name}` option: "
                f"Item `{item}` must contain a key and a value separated by `=`."
            )
        # Split the item into key and value
        key, _, val = item.partition("=")
        key = key.strip()
        if not key:
            # If the key is empty after stripping whitespace, fail with an error message.
            ctx.fail(
                f"Invalid format of `{param.name}` option: Parameter key "
                f"cannot be an empty string."
            )
        # Add "key=value" pair to dot_list.
        dot_list.append(f"{key}={val}")

    conf = OmegaConf.from_dotlist(dot_list)
    return OmegaConf.to_container(conf)


def _split_load_versions(ctx, param, value):
    """Split and format the string coming from the --load-versions
    flag in kedro run, e.g.:
    "dataset1:time1,dataset2:time2" -> {"dataset1": "time1", "dataset2": "time2"}

    Args:
        value: the string with the contents of the --load-versions flag.

    Returns:
        A dictionary with the formatted load versions data.
    """
    if not value:
        return {}

    lv_tuple = tuple(chain.from_iterable(value.split(",") for value in [value]))

    load_versions_dict = {}
    for load_version in lv_tuple:
        load_version = load_version.strip()  # noqa: PLW2901
        load_version_list = load_version.split(":", 1)
        if len(load_version_list) != 2:  # noqa: PLR2004
            raise KedroCliError(
                f"Expected the form of 'load_versions' to be "