Source code for kedro.framework.cli.pipeline

"""A collection of CLI commands for working with Kedro pipelines."""
from __future__ import annotations

import re
import shutil
from pathlib import Path
from textwrap import indent
from typing import Any, NamedTuple

import click

import kedro
from kedro.framework.cli.utils import (
    KedroCliError,
    _clean_pycache,
    command_with_verbosity,
    env_option,
)
from kedro.framework.project import settings
from kedro.framework.startup import ProjectMetadata

_SETUP_PY_TEMPLATE = """# -*- coding: utf-8 -*-
from setuptools import setup, find_packages

setup(
    name="{name}",
    version="{version}",
    description="Modular pipeline `{name}`",
    packages=find_packages(),
    include_package_data=True,
    install_requires={install_requires},
)
"""


[docs] class PipelineArtifacts(NamedTuple): """An ordered collection of source_path, tests_path, config_paths""" pipeline_dir: Path pipeline_tests: Path pipeline_conf: Path
def _assert_pkg_name_ok(pkg_name: str) -> None: """Check that python package name is in line with PEP8 requirements. Args: pkg_name: Candidate Python package name. Raises: KedroCliError: If package name violates the requirements. """ base_message = f"'{pkg_name}' is not a valid Python package name." if not re.match(r"^[a-zA-Z_]", pkg_name): message = base_message + " It must start with a letter or underscore." raise KedroCliError(message) if len(pkg_name) < 2: # noqa: PLR2004 message = base_message + " It must be at least 2 characters long." raise KedroCliError(message) if not re.match(r"^\w+$", pkg_name[1:]): message = ( base_message + " It must contain only letters, digits, and/or underscores." ) raise KedroCliError(message) def _check_pipeline_name(ctx: click.Context, param: Any, value: str) -> str: # noqa: unused-argument if value: _assert_pkg_name_ok(value) return value @click.group(name="Kedro") def pipeline_cli() -> None: # pragma: no cover pass @pipeline_cli.group() def pipeline() -> None: """Commands for working with pipelines.""" @command_with_verbosity(pipeline, "create") @click.argument("name", nargs=1, callback=_check_pipeline_name) @click.option( "--skip-config", is_flag=True, help="Skip creation of config files for the new pipeline(s).", ) @click.option( "template_path", "-t", "--template", type=click.Path(file_okay=False, dir_okay=True, exists=True, path_type=Path), help="Path to cookiecutter template to use for pipeline(s). Will override any local templates.", ) @env_option(help="Environment to create pipeline configuration in. Defaults to `base`.") @click.pass_obj # this will pass the metadata as first argument def create_pipeline( metadata: ProjectMetadata, /, name: str, template_path: Path, skip_config: bool, env: str, **kwargs: Any, ) -> None: # noqa: unused-argument """Create a new modular pipeline by providing a name.""" package_dir = metadata.source_dir / metadata.package_name conf_source = settings.CONF_SOURCE project_conf_path = metadata.project_path / conf_source env = env or "base" if not skip_config and not (project_conf_path / env).exists(): raise KedroCliError( f"Unable to locate environment '{env}'. " f"Make sure it exists in the project configuration." ) # Precedence for template_path is: command line > project templates/pipeline dir > global default # If passed on the CLI, click will verify that the path exists so no need to check again if template_path is None: # No path provided on the CLI, try `PROJECT_PATH/templates/pipeline` template_path = Path(metadata.project_path / "templates" / "pipeline") if not template_path.exists(): # and if that folder doesn't exist fall back to the global default template_path = Path(kedro.__file__).parent / "templates" / "pipeline" click.secho(f"Using pipeline template at: '{template_path}'") result_path = _create_pipeline(name, template_path, package_dir / "pipelines") _copy_pipeline_tests(name, result_path, package_dir) _copy_pipeline_configs(result_path, project_conf_path, skip_config, env=env) click.secho(f"\nPipeline '{name}' was successfully created.\n", fg="green") @command_with_verbosity(pipeline, "delete") @click.argument("name", nargs=1, callback=_check_pipeline_name) @env_option( help="Environment to delete pipeline configuration from. Defaults to 'base'." ) @click.option( "-y", "--yes", is_flag=True, help="Confirm deletion of pipeline non-interactively." ) @click.pass_obj # this will pass the metadata as first argument def delete_pipeline( metadata: ProjectMetadata, /, name: str, env: str, yes: bool, **kwargs: Any ) -> None: # noqa: unused-argument """Delete a modular pipeline by providing a name.""" package_dir = metadata.source_dir / metadata.package_name conf_source = settings.CONF_SOURCE project_conf_path = metadata.project_path / conf_source env = env or "base" if not (project_conf_path / env).exists(): raise KedroCliError( f"Unable to locate environment '{env}'. " f"Make sure it exists in the project configuration." ) pipeline_artifacts = _get_pipeline_artifacts(metadata, pipeline_name=name, env=env) files_to_delete = [ pipeline_artifacts.pipeline_conf / filepath for confdir in ("parameters", "catalog") # Since we remove nesting in 'parameters' and 'catalog' folders, # we want to also del the old project's structure for backward compatibility for filepath in (Path(f"{confdir}_{name}.yml"), Path(confdir) / f"{name}.yml") if (pipeline_artifacts.pipeline_conf / filepath).is_file() ] dirs_to_delete = [ path for path in (pipeline_artifacts.pipeline_dir, pipeline_artifacts.pipeline_tests) if path.is_dir() ] if not files_to_delete and not dirs_to_delete: raise KedroCliError(f"Pipeline '{name}' not found.") if not yes: _echo_deletion_warning( "The following paths will be removed:", directories=dirs_to_delete, files=files_to_delete, ) click.echo() yes = click.confirm(f"Are you sure you want to delete pipeline '{name}'?") click.echo() if not yes: raise KedroCliError("Deletion aborted!") _delete_artifacts(*files_to_delete, *dirs_to_delete) click.secho(f"\nPipeline '{name}' was successfully deleted.", fg="green") click.secho( f"\nIf you added the pipeline '{name}' to 'register_pipelines()' in" f""" '{package_dir / "pipeline_registry.py"}', you will need to remove it.""", fg="yellow", ) def _echo_deletion_warning(message: str, **paths: list[Path]) -> None: paths = {key: values for key, values in paths.items() if values} if paths: click.secho(message, bold=True) for key, values in paths.items(): click.echo(f"\n{key.capitalize()}:") paths_str = "\n".join(str(value) for value in values) click.echo(indent(paths_str, " " * 2)) def _create_pipeline(name: str, template_path: Path, output_dir: Path) -> Path: from cookiecutter.main import cookiecutter cookie_context = {"pipeline_name": name, "kedro_version": kedro.__version__} click.echo(f"Creating the pipeline '{name}': ", nl=False) try: cookiecutter_result = cookiecutter( str(template_path), output_dir=str(output_dir), no_input=True, extra_context=cookie_context, ) except Exception as exc: click.secho("FAILED", fg="red") cls = exc.__class__ raise KedroCliError(f"{cls.__module__}.{cls.__qualname__}: {exc}") from exc click.secho("OK", fg="green") result_path = Path(cookiecutter_result) message = indent(f"Location: '{result_path.resolve()}'", " " * 2) click.secho(message, bold=True) _clean_pycache(result_path) return result_path def _sync_dirs( source: Path, target: Path, prefix: str = "", overwrite: bool = False ) -> None: """Recursively copies `source` directory (or file) into `target` directory without overwriting any existing files/directories in the target using the following rules: 1) Skip any files/directories which names match with files in target, unless overwrite=True. 2) Copy all files from source to target. 3) Recursively copy all directories from source to target. Args: source: A local directory to copy from, must exist. target: A local directory to copy to, will be created if doesn't exist yet. prefix: Prefix for CLI message indentation. """ existing = list(target.iterdir()) if target.is_dir() else [] existing_files = {f.name for f in existing if f.is_file()} existing_folders = {f.name for f in existing if f.is_dir()} if source.is_dir(): content = list(source.iterdir()) elif source.is_file(): content = [source] else: # nothing to copy content = [] # pragma: no cover for source_path in content: source_name = source_path.name target_path = target / source_name click.echo(indent(f"Creating '{target_path}': ", prefix), nl=False) if ( # rule #1 not overwrite and source_name in existing_files or source_path.is_file() and source_name in existing_folders ): click.secho("SKIPPED (already exists)", fg="yellow") elif source_path.is_file(): # rule #2 try: target.mkdir(exist_ok=True, parents=True) shutil.copyfile(str(source_path), str(target_path)) except Exception: click.secho("FAILED", fg="red") raise click.secho("OK", fg="green") else: # source_path is a directory, rule #3 click.echo() new_prefix = (prefix or "") + " " * 2 _sync_dirs(source_path, target_path, prefix=new_prefix) def _get_pipeline_artifacts( project_metadata: ProjectMetadata, pipeline_name: str, env: str ) -> PipelineArtifacts: artifacts = _get_artifacts_to_package( project_metadata, f"pipelines.{pipeline_name}", env ) return PipelineArtifacts(*artifacts) def _get_artifacts_to_package( project_metadata: ProjectMetadata, module_path: str, env: str ) -> tuple[Path, Path, Path]: """From existing project, returns in order: source_path, tests_path, config_paths""" package_dir = project_metadata.source_dir / project_metadata.package_name project_conf_path = project_metadata.project_path / settings.CONF_SOURCE artifacts = ( Path(package_dir, *module_path.split(".")), Path(package_dir.parent, "tests", *module_path.split(".")), project_conf_path / env, ) return artifacts def _copy_pipeline_tests( pipeline_name: str, result_path: Path, package_dir: Path ) -> None: tests_source = result_path / "tests" tests_target = package_dir.parent / "tests" / "pipelines" / pipeline_name try: _sync_dirs(tests_source, tests_target) finally: shutil.rmtree(tests_source) def _copy_pipeline_configs( result_path: Path, conf_path: Path, skip_config: bool, env: str ) -> None: config_source = result_path / "config" try: if not skip_config: config_target = conf_path / env _sync_dirs(config_source, config_target) finally: shutil.rmtree(config_source) def _delete_artifacts(*artifacts: Path) -> None: for artifact in artifacts: click.echo(f"Deleting '{artifact}': ", nl=False) try: if artifact.is_dir(): shutil.rmtree(artifact) else: artifact.unlink() except Exception as exc: click.secho("FAILED", fg="red") cls = exc.__class__ raise KedroCliError(f"{cls.__module__}.{cls.__qualname__}: {exc}") from exc click.secho("OK", fg="green")