Handle secondary-table data path deprecation in the Core API

More specifically, preprocess the data paths in the
`additional_data_tables` argument:

- split each path on "`"
- check that each data path fragment after the split is non-empty
- check that the first fragment is identical to the the name of the
  current dictionary (as specified in `dictionary_name` or
  `train_dictionary_name`)

If all these conditions are met, then convert the legacy data path to
the new format:

- drop the current dictionary name fragment from the beginning of the
  path

- join the remaining data path fragments on "/"

Note: This does not handle external tables, but for the impending beta,
this seems like an acceptable tradeoff.

closes #370

Author: popescu-v

khiops/core/api.py

@@ -320,6 +320,68 @@ def _preprocess_task_arguments(task_args):
             )
         del task_args["max_variable_importances"]
 
+    # Detect deprecated data-path syntax on additional_data_tables
+    if (
+        "additional_data_tables" in task_args
+        and task_args["additional_data_tables"] is not None
+    ):
+        assert "dictionary_name" in task_args or "train_dictionary_name" in task_args
+        if "dictionary_name" in task_args:
+            current_dictionary_name = task_args["dictionary_name"]
+        else:
+            current_dictionary_name = task_args["train_dictionary_name"]
+
+        for kdic_path in set(task_args["additional_data_tables"].keys()):
+            if isinstance(kdic_path, str):
+                deprecated_data_path_separator = "`"
+                data_path_separator = "/"
+                kdic_path_for_warning = kdic_path
+            else:
+                deprecated_data_path_separator = b"`"
+                data_path_separator = b"/"
+                if isinstance(current_dictionary_name, str):
+                    current_dictionary_name = bytes(
+                        current_dictionary_name, encoding="ascii"
+                    )
+                kdic_path_for_warning = kdic_path.decode("ascii")
+
+            # Path split "`" yields non-empty fragments; the first fragment
+            # starts with the current dictionary name
+            kdic_path_parts = kdic_path.split(deprecated_data_path_separator)
+            if all(len(path_part) > 0 for path_part in kdic_path_parts):
+                source_dictionary_name = kdic_path_parts[0]
+
+                if source_dictionary_name == current_dictionary_name:
+                    new_kdic_path_parts = []
+
+                    # Skip current dictionary name from the target path
+                    for kdic_path_part in kdic_path_parts[1:]:
+
+                        # Escape any "/" in the path part
+                        new_kdic_path_parts.append(
+                            kdic_path_part.replace(
+                                data_path_separator,
+                                deprecated_data_path_separator + data_path_separator,
+                            )
+                        )
+
+                    # Replace legacy data path with current data path
+                    new_kdic_path = data_path_separator.join(new_kdic_path_parts)
+                    kdic_file_path = task_args["additional_data_tables"].pop(kdic_path)
+                    task_args["additional_data_tables"][new_kdic_path] = kdic_file_path
+                    warnings.warn(
+                        deprecation_message(
+                            "'`'-based dictionary data path: "
+                            f"'{kdic_path_for_warning}'",
+                            "11.0.1",
+                            replacement=(
+                                "'/'-based dictionary data path "
+                                f"convention: '{new_kdic_path}'"
+                            ),
+                            quote=False,
+                        )
+                    )
+
     # Flatten kwargs
     if "kwargs" in task_args:
         task_args.update(task_args["kwargs"])

tests/test_core.py

@@ -12,6 +12,7 @@
 import textwrap
 import unittest
 import warnings
+from copy import copy
 from pathlib import Path
 from unittest import mock
 
@@ -593,6 +594,119 @@ def test_api_scenario_generation(self):
         # Restore the default runner
         kh.set_runner(default_runner)
 
+    def test_data_path_deprecation_in_api_method(self):
+        """Tests if core.api issues deprecation warning when legacy data
+        paths are used for secondary tables
+        """
+        # Set the root directory of these tests
+        test_resources_dir = os.path.join(resources_dir(), "scenario_generation", "api")
+
+        # Use the test runner that only compares the scenarios
+        default_runner = kh.get_runner()
+        test_runner = ScenarioWriterRunner(self, test_resources_dir)
+        kh.set_runner(test_runner)
+
+        # Obtain mock arguments for each API call
+        method_test_args = self._build_mock_api_method_parameters()
+
+        # Define legacy additional data tables path
+        str_legacy_additional_data_tables = {
+            "Customer`Services": "ServicesBidon.csv",
+            "Customer`Services`Usages": "UsagesBidon.csv",
+            "Customer`Address": "AddressBidon.csv",
+        }
+
+        # Test for each dataset mock parameters
+        for method_name, method_full_args in method_test_args.items():
+            # Use bytes for deploy_model's additional_data_tables
+            if method_name == "deploy_model":
+                legacy_additional_data_tables = {
+                    bytes(key, encoding="ascii"): bytes(value, encoding="ascii")
+                    for key, value in str_legacy_additional_data_tables.items()
+                }
+            else:
+                legacy_additional_data_tables = str_legacy_additional_data_tables
+            # Set the runners test name
+            test_runner.test_name = method_name
+
+            # Clean the directory for this method's tests
+            cleanup_dir(test_runner.output_scenario_dir, "*/output/*._kh", verbose=True)
+            for dataset, dataset_method_args in method_full_args.items():
+                # Test only for the Customer dataset
+                if dataset != "Customer":
+                    continue
+
+                test_runner.subtest_name = dataset
+                with self.subTest(method=method_name):
+                    # Get the API function and its args and kwargs
+                    method = getattr(kh, method_name)
+                    dataset_args = dataset_method_args["args"]
+                    dataset_kwargs = dataset_method_args["kwargs"]
+
+                    # Skip the test if `additional_data_tables` is not an
+                    # API call kwarg
+                    if "additional_data_tables" not in dataset_kwargs:
+                        continue
+
+                    # Store current additional data_tables
+                    current_additional_data_tables = copy(
+                        dataset_kwargs["additional_data_tables"]
+                    )
+
+                    # Update the `additional_data_tables` kwargs to use
+                    # legacy paths
+                    dataset_kwargs["additional_data_tables"] = copy(
+                        legacy_additional_data_tables
+                    )
+
+                    # Test that using legacy paths entails a deprecation warning
+                    with warnings.catch_warnings(record=True) as warning_list:
+                        method(*dataset_args, **dataset_kwargs)
+
+                    # Build current-legacy data path map
+                    legacy_to_current_data_paths = {}
+                    for (
+                        data_path,
+                        data_file_path,
+                    ) in current_additional_data_tables.items():
+                        for (
+                            leg_data_path,
+                            leg_data_file_path,
+                        ) in legacy_additional_data_tables.items():
+                            if leg_data_file_path == data_file_path:
+                                legacy_to_current_data_paths[leg_data_path] = data_path
+                                break
+
+                    # Check the warning message
+                    self.assertEqual(
+                        len(warning_list), len(legacy_additional_data_tables)
+                    )
+                    warning = warning_list[0]
+                    for warning in warning_list:
+                        self.assertTrue(issubclass(warning.category, UserWarning))
+                        warning_message = warning.message
+                        self.assertEqual(len(warning_message.args), 1)
+                        message = warning_message.args[0]
+                        self.assertTrue(
+                            "'`'-based dictionary data path" in message
+                            and "deprecated" in message
+                        )
+
+                        # Check legacy data path is replaced with the current
+                        # data path
+                        for legacy_data_path in legacy_additional_data_tables:
+                            expected_legacy_data_path = legacy_to_current_data_paths[
+                                legacy_data_path
+                            ]
+                            if f"'{legacy_data_path}'" in message:
+                                self.assertTrue(
+                                    f"'{expected_legacy_data_path}'" in message
+                                )
+                                break
+
+        # Restore the default runner
+        kh.set_runner(default_runner)
+
     def test_unknown_argument_in_api_method(self):
         """Tests if core.api raises ValueError when an unknown argument is passed"""
         # Obtain mock arguments for each API call