CLI Utilities API Reference

from kedro.framework.cli import main, command_with_verbosity, load_entry_points
from kedro.framework.cli.utils import (
    call, python_call, forward_command,
    validate_conf_source, split_string, split_node_names,
    env_option, find_run_command
)

def main() -> None:
    """
    Main entry point for Kedro CLI.
    Entry point command: `kedro`
    """

def load_obj(obj_path: str, default_obj_path: str = "") -> Any:
    """
    Extract an object from a given path.

    Parameters:
    - obj_path: Path to object including name (e.g., 'module.submodule.ClassName')
    - default_obj_path: Default object path if obj_path is incomplete

    Returns:
    Extracted object (class, function, etc.)
    """

def is_kedro_project(project_path: str | Path) -> bool:
    """
    Check if a path is the root directory of a Kedro project.

    Parameters:
    - project_path: Path to check

    Returns:
    True if path is a Kedro project root
    """

def find_kedro_project(current_dir: Path) -> Path | None:
    """
    Find Kedro project associated with a given path.

    Parameters:
    - current_dir: Starting directory for search

    Returns:
    Path to Kedro project root, or None if not found
    """

def command_with_verbosity(func: Callable) -> Callable:
    """
    Decorator to add verbosity option to Click commands.

    Parameters:
    - func: Click command function

    Returns:
    Decorated function with --verbose/-v option

    Example:
    >>> @click.command()
    ... @command_with_verbosity
    ... def my_command(verbose):
    ...     if verbose:
    ...         print("Verbose mode enabled")
    """

def load_entry_points(group: str) -> dict[str, Callable]:
    """
    Load and return entry point groups for extending CLI.

    Parameters:
    - group: Entry point group name

    Returns:
    Dictionary mapping entry point names to callables
    """

def call(cmd: list[str], **kwargs: Any) -> None:
    """
    Run a subprocess command and raise if it fails.

    Parameters:
    - cmd: List of command parts (e.g., ['git', 'status'])
    - **kwargs: Optional keyword arguments passed to subprocess.run

    Raises:
    - click.exceptions.Exit: If subprocess.run returns non-zero code

    Example:
    >>> from kedro.framework.cli.utils import call
    >>> call(['git', 'status'])
    """

def python_call(module: str, arguments: Iterable[str], **kwargs: Any) -> None:
    """
    Run a subprocess command that invokes a Python module.

    Parameters:
    - module: Python module name to invoke (e.g., 'pip')
    - arguments: Arguments to pass to the module
    - **kwargs: Optional keyword arguments passed to subprocess.run

    Example:
    >>> from kedro.framework.cli.utils import python_call
    >>> python_call('pip', ['install', 'kedro'])
    """

def forward_command(group: Any, name: str | None = None, forward_help: bool = False) -> Callable:
    """
    Decorator that creates a Click command receiving the rest of the command line as 'args'.

    Parameters:
    - group: Click group to add command to
    - name: Command name (defaults to function name)
    - forward_help: If True, forwards help flag to wrapped command

    Returns:
    Decorator function

    Example:
    >>> @forward_command(cli, name="docker")
    ... def docker_cmd(args):
    ...     # args contains all remaining CLI arguments
    ...     pass
    """

def split_string(ctx: click.Context, param: Any, value: str) -> list[str]:
    """
    Split string by comma (Click callback).

    Parameters:
    - ctx: Click context
    - param: Click parameter
    - value: String to split

    Returns:
    List of stripped string items

    Example:
    >>> split_string(ctx, param, "tag1, tag2, tag3")
    ['tag1', 'tag2', 'tag3']
    """

def split_node_names(ctx: click.Context, param: Any, to_split: str) -> list[str]:
    """
    Split string by comma, ignoring commas enclosed by square brackets.
    Used for parsing node names which may contain commas in their default names.

    Parameters:
    - ctx: Click context
    - param: Click parameter
    - to_split: String to split safely

    Returns:
    List of node names

    Example:
    >>> split_node_names(ctx, param, "node1,func([a,b]) -> [c],node2")
    ['node1', 'func([a,b]) -> [c]', 'node2']
    """

def validate_conf_source(ctx: click.Context, param: Any, value: str) -> str | None:
    """
    Validate the conf_source, only checking existence for local paths.
    Remote URLs (except file://) are passed through without validation.

    Parameters:
    - ctx: Click context
    - param: Click parameter
    - value: Configuration source path or URL

    Returns:
    Resolved path string or None if value is empty

    Raises:
    - click.BadParameter: If local path does not exist or is invalid

    Example:
    >>> validate_conf_source(ctx, param, "conf")
    '/absolute/path/to/conf'
    >>> validate_conf_source(ctx, param, "s3://bucket/conf")
    's3://bucket/conf'
    """

def env_option(func_: Callable | None = None, **kwargs: Any) -> Callable:
    """
    Add --env CLI option to a function.
    Decorator to add environment selection option to Click commands.

    Parameters:
    - func_: Function to decorate (optional)
    - **kwargs: Additional arguments for click.option (type, default, help, etc.)

    Returns:
    Decorated function or decorator

    Example:
    >>> @env_option
    ... def my_command(env):
    ...     print(f"Using environment: {env}")
    """

def find_run_command(package_name: str) -> Callable:
    """
    Find the run command to be executed.
    This is either the default run command defined in the Kedro framework
    or a run command defined by an installed plugin.

    Parameters:
    - package_name: The name of the package being run

    Returns:
    Run command callable to be executed

    Raises:
    - KedroCliError: If the run command is not found

    Example:
    >>> from kedro.framework.cli.utils import find_run_command
    >>> run_cmd = find_run_command("my_kedro_project")
    """

ORANGE: tuple[int, int, int]  # RGB (255, 175, 0) for CLI styling
BRIGHT_BLACK: tuple[int, int, int]  # RGB (128, 128, 128) for CLI styling

from kedro.framework.cli.utils import CommandCollection, LazyGroup, KedroCliError

class CommandCollection(click.CommandCollection):
    """
    Modified Click CommandCollection that supports grouping commands.

    Extends click.CommandCollection to organize commands into titled groups
    while still running the source groups' functionality.

    Parameters:
    - *groups: Tuples of (title, list of click.Group instances)

    Example:
    >>> from click import Group
    >>> from kedro.framework.cli.utils import CommandCollection
    >>>
    >>> global_group = Group("global")
    >>> project_group = Group("project")
    >>>
    >>> cli = CommandCollection(
    ...     ("Global commands", [global_group]),
    ...     ("Project commands", [project_group])
    ... )
    """

class LazyGroup(click.Group):
    """
    Click Group that supports lazy loading of subcommands.

    Subcommands are only imported and loaded when accessed, improving
    CLI startup time for large projects with many commands.

    Parameters:
    - lazy_subcommands: Dictionary mapping command names to import paths
        Format: {"command_name": "module.path:function_name"}

    Example:
    >>> from kedro.framework.cli.utils import LazyGroup
    >>>
    >>> @LazyGroup(
    ...     lazy_subcommands={
    ...         "run": "kedro.framework.cli.project:run",
    ...         "catalog": "kedro.framework.cli.catalog:catalog"
    ...     }
    ... )
    ... def cli():
    ...     pass
    """

class KedroCliError(click.exceptions.ClickException):
    """
    Custom exception for Kedro CLI errors.

    Extends Click's ClickException to provide Kedro-specific error handling
    and formatting. Users should pass an appropriate error message at construction.

    Attributes:
    - VERBOSE_ERROR: bool - Whether to show verbose error output (default: False)
    - VERBOSE_EXISTS: bool - Whether verbose mode is available (default: True)
    - COOKIECUTTER_EXCEPTIONS_PREFIX: str - Prefix for cookiecutter exceptions

    Raises:
    Common scenarios include:
    - Invalid CLI command or arguments
    - Failed command execution
    - Missing required parameters
    - Plugin loading failures

    Example:
    >>> from kedro.framework.cli.utils import KedroCliError
    >>>
    >>> if not project_path.exists():
    ...     raise KedroCliError("Project path does not exist")
    """

from kedro.utils import load_obj

# Load a dataset class dynamically
DatasetClass = load_obj("kedro.io.MemoryDataset")
dataset = DatasetClass()

# Load custom class
CustomClass = load_obj("my_package.custom_module.CustomClass")

from pathlib import Path
from kedro.utils import is_kedro_project, find_kedro_project

# Check if current directory is a Kedro project
if is_kedro_project(Path.cwd()):
    print("This is a Kedro project!")

# Find nearest Kedro project
project_path = find_kedro_project(Path.cwd())
if project_path:
    print(f"Found Kedro project at: {project_path}")