[KED-2503] Simplify template cli.py (kedro-org#1041)

Author: Lorena Bălan

features/steps/test_starter/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/cli.py

@@ -32,24 +32,22 @@
 import logging
 from itertools import chain
 from pathlib import Path
-from typing import Dict, Iterable, Tuple
+from typing import Iterable, Tuple
 
 import click
-from kedro.framework.cli.catalog import catalog as catalog_group
-from kedro.framework.cli.jupyter import jupyter as jupyter_group
-from kedro.framework.cli.pipeline import pipeline as pipeline_group
-from kedro.framework.cli.project import project_group
-from kedro.framework.cli.utils import KedroCliError, env_option, split_string
+from kedro.framework.cli.utils import (
+    KedroCliError,
+    _config_file_callback,
+    _reformat_load_versions,
+    _split_params,
+    env_option,
+    split_string,
+)
 from kedro.framework.session import KedroSession
 from kedro.utils import load_obj
 
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
-# get our package onto the python path
-PROJ_PATH = Path(__file__).resolve().parent
-
-ENV_ARG_HELP = """Run the pipeline in a configured environment. If not specified,
-pipeline will run using environment `local`."""
 FROM_INPUTS_HELP = (
     """A list of dataset names which should be used as a starting point."""
 )
@@ -79,79 +77,10 @@
 so parameter values are allowed to contain colons, parameter keys are not."""
 
 
-def _config_file_callback(ctx, param, value):  # pylint: disable=unused-argument
-    """Config file callback, that replaces command line options with config file
-    values. If command line options are passed, they override config file values.
-    """
-    # for performance reasons
-    import anyconfig  # pylint: disable=import-outside-toplevel
-
-    ctx.default_map = ctx.default_map or {}
-    section = ctx.info_name
-
-    if value:
-        config = anyconfig.load(value)[section]
-        ctx.default_map.update(config)
-
-    return value
-
-
 def _get_values_as_tuple(values: Iterable[str]) -> Tuple[str, ...]:
     return tuple(chain.from_iterable(value.split(",") for value in values))
 
 
-def _reformat_load_versions(  # pylint: disable=unused-argument
-    ctx, param, value
-) -> Dict[str, str]:
-    """Reformat data structure from tuple to dictionary for `load-version`, e.g:
-    ('dataset1:time1', 'dataset2:time2') -> {"dataset1": "time1", "dataset2": "time2"}.
-    """
-    load_versions_dict = {}
-
-    for load_version in value:
-        load_version_list = load_version.split(":", 1)
-        if len(load_version_list) != 2:
-            raise KedroCliError(
-                f"Expected the form of `load_version` to be "
-                f"`dataset_name:YYYY-MM-DDThh.mm.ss.sssZ`,"
-                f"found {load_version} instead"
-            )
-        load_versions_dict[load_version_list[0]] = load_version_list[1]
-
-    return load_versions_dict
-
-
-def _split_params(ctx, param, value):
-    if isinstance(value, dict):
-        return value
-    result = {}
-    for item in split_string(ctx, param, value):
-        item = item.split(":", 1)
-        if len(item) != 2:
-            ctx.fail(
-                f"Invalid format of `{param.name}` option: "
-                f"Item `{item[0]}` must contain "
-                f"a key and a value separated by `:`."
-            )
-        key = item[0].strip()
-        if not key:
-            ctx.fail(
-                f"Invalid format of `{param.name}` option: Parameter key "
-                f"cannot be an empty string."
-            )
-        value = item[1].strip()
-        result[key] = _try_convert_to_numeric(value)
-    return result
-
-
-def _try_convert_to_numeric(value):
-    try:
-        value = float(value)
-    except ValueError:
-        return value
-    return int(value) if value.is_integer() else value
-
-
 @click.group(context_settings=CONTEXT_SETTINGS, name=__file__)
 def cli():
     """Command line tools for manipulating a Kedro project."""
@@ -239,11 +168,3 @@ def run(
         # Logging parameters for some e2e tests
         params_to_log = session.load_context().params
         logging.info("Parameters: %s", json.dumps(params_to_log, sort_keys=True))
-
-
-cli.add_command(pipeline_group)
-cli.add_command(catalog_group)
-cli.add_command(jupyter_group)
-
-for command in project_group.commands.values():
-    cli.add_command(command)

kedro/framework/cli/utils.py

@@ -41,7 +41,7 @@
 from importlib import import_module
 from itertools import chain
 from pathlib import Path
-from typing import Iterable, List, Mapping, Sequence, Set, Tuple, Union
+from typing import Dict, Iterable, List, Mapping, Sequence, Set, Tuple, Union
 
 import click
 import pkg_resources
@@ -396,3 +396,73 @@ def _add_src_to_path(source_dir: Path, project_path: Path) -> None:  # pragma: n
     )
     warnings.warn(msg, FutureWarning)
     real_add_src_to_path(source_dir, project_path)
+
+
+def _config_file_callback(ctx, param, value):  # pylint: disable=unused-argument
+    """CLI callback that replaces command line options
+    with values specified in a config file. If command line
+    options are passed, they override config file values.
+    """
+    # for performance reasons
+    import anyconfig  # pylint: disable=import-outside-toplevel
+
+    ctx.default_map = ctx.default_map or {}
+    section = ctx.info_name
+
+    if value:
+        config = anyconfig.load(value)[section]
+        ctx.default_map.update(config)
+
+    return value
+
+
+def _reformat_load_versions(  # pylint: disable=unused-argument
+    ctx, param, value
+) -> Dict[str, str]:
+    """Reformat data structure from tuple to dictionary for `load-version`, e.g.:
+    ('dataset1:time1', 'dataset2:time2') -> {"dataset1": "time1", "dataset2": "time2"}.
+    """
+    load_versions_dict = {}
+
+    for load_version in value:
+        load_version_list = load_version.split(":", 1)
+        if len(load_version_list) != 2:
+            raise KedroCliError(
+                f"Expected the form of `load_version` to be "
+                f"`dataset_name:YYYY-MM-DDThh.mm.ss.sssZ`,"
+                f"found {load_version} instead"
+            )
+        load_versions_dict[load_version_list[0]] = load_version_list[1]
+
+    return load_versions_dict
+
+
+def _try_convert_to_numeric(value):
+    try:
+        value = float(value)
+    except ValueError:
+        return value
+    return int(value) if value.is_integer() else value
+
+
+def _split_params(ctx, param, value):
+    if isinstance(value, dict):
+        return value
+    result = {}
+    for item in split_string(ctx, param, value):
+        item = item.split(":", 1)
+        if len(item) != 2:
+            ctx.fail(
+                f"Invalid format of `{param.name}` option: "
+                f"Item `{item[0]}` must contain "
+                f"a key and a value separated by `:`."
+            )
+        key = item[0].strip()
+        if not key:
+            ctx.fail(
+                f"Invalid format of `{param.name}` option: Parameter key "
+                f"cannot be an empty string."
+            )
+        value = item[1].strip()
+        result[key] = _try_convert_to_numeric(value)
+    return result

kedro/templates/project/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/cli.py

@@ -30,24 +30,22 @@
 Intended to be invoked via `kedro`."""
 from itertools import chain
 from pathlib import Path
-from typing import Dict, Iterable, Tuple
+from typing import Iterable, Tuple
 
 import click
-from kedro.framework.cli.catalog import catalog as catalog_group
-from kedro.framework.cli.jupyter import jupyter as jupyter_group
-from kedro.framework.cli.pipeline import pipeline as pipeline_group
-from kedro.framework.cli.project import project_group
-from kedro.framework.cli.utils import KedroCliError, env_option, split_string
+from kedro.framework.cli.utils import (
+    KedroCliError,
+    _config_file_callback,
+    _reformat_load_versions,
+    _split_params,
+    env_option,
+    split_string,
+)
 from kedro.framework.session import KedroSession
 from kedro.utils import load_obj
 
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
-# get our package onto the python path
-PROJ_PATH = Path(__file__).resolve().parent
-
-ENV_ARG_HELP = """Run the pipeline in a configured environment. If not specified,
-pipeline will run using environment `local`."""
 FROM_INPUTS_HELP = (
     """A list of dataset names which should be used as a starting point."""
 )
@@ -78,79 +76,10 @@
 so parameter values are allowed to contain colons, parameter keys are not."""
 
 
-def _config_file_callback(ctx, param, value):  # pylint: disable=unused-argument
-    """Config file callback, that replaces command line options with config file
-    values. If command line options are passed, they override config file values.
-    """
-    # for performance reasons
-    import anyconfig  # pylint: disable=import-outside-toplevel
-
-    ctx.default_map = ctx.default_map or {}
-    section = ctx.info_name
-
-    if value:
-        config = anyconfig.load(value)[section]
-        ctx.default_map.update(config)
-
-    return value
-
-
 def _get_values_as_tuple(values: Iterable[str]) -> Tuple[str, ...]:
     return tuple(chain.from_iterable(value.split(",") for value in values))
 
 
-def _reformat_load_versions(  # pylint: disable=unused-argument
-    ctx, param, value
-) -> Dict[str, str]:
-    """Reformat data structure from tuple to dictionary for `load-version`, e.g.:
-    ('dataset1:time1', 'dataset2:time2') -> {"dataset1": "time1", "dataset2": "time2"}.
-    """
-    load_versions_dict = {}
-
-    for load_version in value:
-        load_version_list = load_version.split(":", 1)
-        if len(load_version_list) != 2:
-            raise KedroCliError(
-                f"Expected the form of `load_version` to be "
-                f"`dataset_name:YYYY-MM-DDThh.mm.ss.sssZ`,"
-                f"found {load_version} instead"
-            )
-        load_versions_dict[load_version_list[0]] = load_version_list[1]
-
-    return load_versions_dict
-
-
-def _split_params(ctx, param, value):
-    if isinstance(value, dict):
-        return value
-    result = {}
-    for item in split_string(ctx, param, value):
-        item = item.split(":", 1)
-        if len(item) != 2:
-            ctx.fail(
-                f"Invalid format of `{param.name}` option: "
-                f"Item `{item[0]}` must contain "
-                f"a key and a value separated by `:`."
-            )
-        key = item[0].strip()
-        if not key:
-            ctx.fail(
-                f"Invalid format of `{param.name}` option: Parameter key "
-                f"cannot be an empty string."
-            )
-        value = item[1].strip()
-        result[key] = _try_convert_to_numeric(value)
-    return result
-
-
-def _try_convert_to_numeric(value):
-    try:
-        value = float(value)
-    except ValueError:
-        return value
-    return int(value) if value.is_integer() else value
-
-
 @click.group(context_settings=CONTEXT_SETTINGS, name=__file__)
 def cli():
     """Command line tools for manipulating a Kedro project."""
@@ -239,11 +168,3 @@ def run(
             load_versions=load_version,
             pipeline_name=pipeline,
         )
-
-
-cli.add_command(pipeline_group)
-cli.add_command(catalog_group)
-cli.add_command(jupyter_group)
-
-for command in project_group.commands.values():
-    cli.add_command(command)

test_requirements.txt

@@ -14,7 +14,7 @@ holoviews~=1.13.0
 import-linter==1.0
 ipython==7.10  # Due to https://github.com/ipython/ipykernel/issues/540
 joblib>=0.14
-matplotlib>=3.0.3, <4.0
+matplotlib>=3.0.3, <3.4  # 3.4.0 breaks holoviews
 memory_profiler>=0.50.0, <1.0
 moto==1.3.7
 mypy<=1.0

tests/framework/cli/conftest.py

@@ -43,7 +43,11 @@
 from pytest import fixture
 
 from kedro import __version__ as kedro_version
+from kedro.framework.cli.catalog import catalog_cli
 from kedro.framework.cli.cli import cli
+from kedro.framework.cli.jupyter import jupyter_cli
+from kedro.framework.cli.pipeline import pipeline_cli
+from kedro.framework.cli.project import project_group
 from kedro.framework.cli.starters import create_cli
 from kedro.framework.project import configure_project, pipelines, settings
 from kedro.framework.startup import ProjectMetadata
@@ -114,7 +118,17 @@ def fake_metadata(fake_root_dir):
 # code when invoking `kedro` but when imported, they still need to be merged
 @fixture(scope="module")
 def fake_kedro_cli():
-    return click.CommandCollection(name="Kedro", sources=[cli, create_cli])
+    return click.CommandCollection(
+        name="Kedro",
+        sources=[
+            cli,
+            create_cli,
+            catalog_cli,
+            jupyter_cli,
+            pipeline_cli,
+            project_group,
+        ],
+    )
 
 
 @fixture(scope="module")
@@ -125,8 +139,7 @@ def fake_project_cli(
     starter_path = Path(__file__).parents[3].resolve()
     starter_path = starter_path / "features" / "steps" / "test_starter"
     CliRunner().invoke(
-        fake_kedro_cli,
-        ["new", "-c", str(dummy_config), "--starter", str(starter_path)],
+        fake_kedro_cli, ["new", "-c", str(dummy_config), "--starter", str(starter_path)]
     )
 
     # NOTE: Here we load a couple of modules, as they would be imported in
@@ -138,7 +151,7 @@ def fake_project_cli(
 
     import_module(PACKAGE_NAME)
     configure_project(PACKAGE_NAME)
-    yield getattr(import_module(f"{PACKAGE_NAME}.cli"), "cli")
+    yield fake_kedro_cli
 
     # reset side-effects of configure_project
     pipelines.clear()