CLI Hooks API Reference

from kedro.framework.cli.hooks import (
    cli_hook_impl,
    CLIHooksManager,
    get_cli_hook_manager
)
from kedro.framework.cli.hooks.markers import CLI_HOOK_NAMESPACE, cli_hook_spec

CLI_HOOK_NAMESPACE: str  # "kedro_cli" - Namespace identifier for CLI hooks

cli_hook_spec: HookspecMarker  # Decorator for marking CLI hook specifications
    """
    Marker for CLI hook specification methods.
    Used internally to define CLI hook specifications.

    Example:
    >>> from kedro.framework.cli.hooks.markers import cli_hook_spec
    >>> class MyCLIHookSpecs:
    ...     @cli_hook_spec
    ...     def my_cli_hook(self, arg1, arg2):
    ...         pass
    """

def cli_hook_impl(func: Callable) -> Callable:
    """
    Decorator to mark methods as CLI hook implementations.

    Similar to hook_impl but specifically for CLI lifecycle events.

    Parameters:
    - func: Method to mark as CLI hook implementation

    Returns:
    Decorated function registered as CLI hook implementation

    Example:
    >>> from kedro.framework.cli.hooks import cli_hook_impl
    >>> class MyCLIHooks:
    ...     @cli_hook_impl
    ...     def before_command_run(self, command_args):
    ...         print(f"Running: {' '.join(command_args)}")
    """

class CLIHooksManager:
    """
    Manager for CLI-specific hooks.
    Handles registration and execution of CLI lifecycle hooks.
    """

    def register(self, plugin: Any) -> None:
        """
        Register a CLI hooks plugin.

        Parameters:
        - plugin: Hook class instance to register
        """

    def hook(self) -> Any:
        """
        Get hook caller for executing registered hooks.

        Returns:
        Hook caller object
        """

def get_cli_hook_manager() -> CLIHooksManager:
    """
    Get the CLI hook manager instance.

    Returns:
    CLIHooksManager for CLI hook operations

    Example:
    >>> manager = get_cli_hook_manager()
    >>> manager.register(MyCLIHooks())
    """

class CLICommandSpecs:
    """
    Hook specifications for CLI command lifecycle events.
    """

    @hook_spec
    def before_command_run(
        self,
        project_metadata: ProjectMetadata,
        command_args: list[str]
    ) -> None:
        """
        Hook called before a CLI command runs.

        Parameters:
        - project_metadata: Project metadata information
        - command_args: Command-line arguments

        Example:
        >>> @cli_hook_impl
        >>> def before_command_run(self, command_args, **kwargs):
        ...     print(f"Executing: {' '.join(command_args)}")
        """

    @hook_spec
    def after_command_run(
        self,
        project_metadata: ProjectMetadata,
        command_args: list[str],
        exit_code: int
    ) -> None:
        """
        Hook called after a CLI command completes.

        Parameters:
        - project_metadata: Project metadata information
        - command_args: Command-line arguments
        - exit_code: Command exit code (0 for success)

        Example:
        >>> @cli_hook_impl
        >>> def after_command_run(self, exit_code, **kwargs):
        ...     if exit_code == 0:
        ...         print("Success!")
        ...     else:
        ...         print(f"Failed with code {exit_code}")
        """

from kedro.framework.cli.hooks import cli_hook_impl

class MyCLIHooks:
    """CLI hooks for command tracking."""

    @cli_hook_impl
    def before_command_run(self, command_args):
        """Log command execution."""
        print(f"Executing: {' '.join(command_args)}")

    @cli_hook_impl
    def after_command_run(self, command_args, exit_code):
        """Log command completion."""
        if exit_code == 0:
            print("Command completed successfully")
        else:
            print(f"Command failed with exit code {exit_code}")

import time
from kedro.framework.cli.hooks import cli_hook_impl

class TimingCLIHooks:
    """Measure CLI command execution time."""

    def __init__(self):
        self.start_time = None

    @cli_hook_impl
    def before_command_run(self, command_args):
        """Record start time."""
        self.start_time = time.time()
        print(f"Starting: {' '.join(command_args)}")

    @cli_hook_impl
    def after_command_run(self, command_args, exit_code):
        """Report execution time."""
        duration = time.time() - self.start_time
        print(f"Completed in {duration:.2f} seconds")

import logging
from kedro.framework.cli.hooks import cli_hook_impl

logger = logging.getLogger(__name__)

class LoggingCLIHooks:
    """Log all CLI commands."""

    @cli_hook_impl
    def before_command_run(self, project_metadata, command_args):
        """Log command start."""
        logger.info(
            f"Project: {project_metadata.project_name}, "
            f"Command: {' '.join(command_args)}"
        )

    @cli_hook_impl
    def after_command_run(self, exit_code):
        """Log command result."""
        if exit_code == 0:
            logger.info("Command succeeded")
        else:
            logger.error(f"Command failed with exit code {exit_code}")

# Register CLI hooks
from kedro.framework.cli.hooks import get_cli_hook_manager

manager = get_cli_hook_manager()
manager.register(MyCLIHooks())
manager.register(TimingCLIHooks())
manager.register(LoggingCLIHooks())