"""kedro is a CLI for managing Kedro projects.
This module implements commands available from the kedro CLI.
"""
from __future__ import annotations
import importlib
import sys
import traceback
from collections import defaultdict
from pathlib import Path
from typing import Any, Sequence
import click
from kedro import __version__ as version
from kedro.framework.cli import BRIGHT_BLACK, ORANGE
from kedro.framework.cli.catalog import catalog_cli
from kedro.framework.cli.hooks import get_cli_hook_manager
from kedro.framework.cli.jupyter import jupyter_cli
from kedro.framework.cli.micropkg import micropkg_cli
from kedro.framework.cli.pipeline import pipeline_cli
from kedro.framework.cli.project import project_group
from kedro.framework.cli.registry import registry_cli
from kedro.framework.cli.starters import create_cli
from kedro.framework.cli.utils import (
CONTEXT_SETTINGS,
ENTRY_POINT_GROUPS,
CommandCollection,
KedroCliError,
_get_entry_points,
load_entry_points,
)
from kedro.framework.project import LOGGING # noqa: F401
from kedro.framework.startup import bootstrap_project
from kedro.utils import _find_kedro_project, _is_project
LOGO = rf"""
_ _
| | _____ __| |_ __ ___
| |/ / _ \/ _` | '__/ _ \
| < __/ (_| | | | (_) |
|_|\_\___|\__,_|_| \___/
v{version}
"""
@click.group(context_settings=CONTEXT_SETTINGS, name="Kedro")
@click.version_option(version, "--version", "-V", help="Show version and exit")
def cli() -> None: # pragma: no cover
"""Kedro is a CLI for creating and using Kedro projects. For more
information, type ``kedro info``.
"""
pass
@cli.command()
def info() -> None:
"""Get more information about kedro."""
click.secho(LOGO, fg="green")
click.echo(
"Kedro is a Python framework for\n"
"creating reproducible, maintainable\n"
"and modular data science code."
)
plugin_versions = {}
plugin_entry_points = defaultdict(set)
for plugin_entry_point in ENTRY_POINT_GROUPS:
for entry_point in _get_entry_points(plugin_entry_point):
module_name = entry_point.module.split(".")[0]
plugin_versions[module_name] = entry_point.dist.version
plugin_entry_points[module_name].add(plugin_entry_point)
click.echo()
if plugin_versions:
click.echo("Installed plugins:")
for plugin_name, plugin_version in sorted(plugin_versions.items()):
entrypoints_str = ",".join(sorted(plugin_entry_points[plugin_name]))
click.echo(
f"{plugin_name}: {plugin_version} (entry points:{entrypoints_str})"
)
else:
click.echo("No plugins installed")
def _init_plugins() -> None:
init_hooks = load_entry_points("init")
for init_hook in init_hooks:
init_hook()
[docs]
class KedroCLI(CommandCollection):
"""A CommandCollection class to encapsulate the KedroCLI command
loading.
"""
def __init__(self, project_path: Path):
self._metadata = None # running in package mode
if _is_project(project_path):
self._metadata = bootstrap_project(project_path)
self._cli_hook_manager = get_cli_hook_manager()
super().__init__(
("Global commands", self.global_groups),
("Project specific commands", self.project_groups),
)
def main(
self,
args: Any | None = None,
prog_name: Any | None = None,
complete_var: Any | None = None,
standalone_mode: bool = True,
**extra: Any,
) -> Any:
if self._metadata:
extra.update(obj=self._metadata)
# This is how click's internals parse sys.argv, which include the command,
# subcommand, arguments and options. click doesn't store this information anywhere
# so we have to re-do it.
args = sys.argv[1:] if args is None else list(args)
self._cli_hook_manager.hook.before_command_run(
project_metadata=self._metadata, command_args=args
)
try:
super().main(
args=args,
prog_name=prog_name,
complete_var=complete_var,
standalone_mode=standalone_mode,
**extra,
)
# click.core.main() method exits by default, we capture this and then
# exit as originally intended
except SystemExit as exc:
self._cli_hook_manager.hook.after_command_run(
project_metadata=self._metadata, command_args=args, exit_code=exc.code
)
# When CLI is run outside of a project, project_groups are not registered
catch_exception = "click.exceptions.UsageError: No such command"
# click convert exception handles to error message
if catch_exception in traceback.format_exc() and not self.project_groups:
warn = click.style(
"\nKedro project not found in this directory. ",
fg=ORANGE,
bold=True,
)
result = (
click.style("Project specific commands such as ")
+ click.style("'run' ", fg="cyan")
+ "or "
+ click.style("'jupyter' ", fg="cyan")
+ "are only available within a project directory."
)
message = warn + result
hint = (
click.style(
"\nHint: Kedro is looking for a file called ", fg=BRIGHT_BLACK
)
+ click.style("'pyproject.toml", fg="magenta")
+ click.style(
", is one present in your current working directory?",
fg=BRIGHT_BLACK,
)
)
click.echo(message)
click.echo(hint)
sys.exit(exc.code)
@property
def global_groups(self) -> Sequence[click.MultiCommand]:
"""Property which loads all global command groups from plugins and
combines them with the built-in ones (eventually overriding the
built-in ones if they are redefined by plugins).
"""
return [cli, create_cli, *load_entry_points("global")]
@property
def project_groups(self) -> Sequence[click.MultiCommand]:
"""Property which loads all project command groups from the
project and the plugins, then combines them with the built-in ones.
Built-in commands can be overridden by plugins, which can be
overridden by a custom project cli.py.
See https://kedro.readthedocs.io/en/stable/extend_kedro/common_use_cases.html#use-case-3-how-to-add-or-modify-cli-commands
on how to add this.
"""
if not self._metadata:
return []
built_in = [
catalog_cli,
jupyter_cli,
pipeline_cli,
micropkg_cli,
project_group,
registry_cli,
]
plugins = load_entry_points("project")
try:
project_cli = importlib.import_module(f"{self._metadata.package_name}.cli")
# fail gracefully if cli.py does not exist
except ModuleNotFoundError:
# return only built-in commands and commands from plugins
# (plugins can override built-in commands)
return [*built_in, *plugins]
# fail badly if cli.py exists, but has no `cli` in it
if not hasattr(project_cli, "cli"):
raise KedroCliError(
f"Cannot load commands from {self._metadata.package_name}.cli"
)
user_defined = project_cli.cli
# return built-in commands, plugin commands and user defined commands
# (overriding happens as follows built-in < plugins < cli.py)
return [*built_in, *plugins, user_defined]
[docs]
def main() -> None: # pragma: no cover
"""Main entry point. Look for a ``cli.py``, and, if found, add its
commands to `kedro`'s before invoking the CLI.
"""
_init_plugins()
cli_collection = KedroCLI(
project_path=_find_kedro_project(Path.cwd()) or Path.cwd()
)
cli_collection()
