IPython Extension API Reference

from types import MappingProxyType
from typing import Final

FunctionParameters = MappingProxyType
# Type alias for function parameter mappings used in node inspection

RICH_INSTALLED: Final[bool]
# Boolean constant indicating whether the Rich library is installed and available
# Used internally to determine if Rich formatting features can be enabled

def load_ipython_extension(ipython: Any) -> None:
    """
    Load Kedro IPython extension.

    Provides magic commands and utilities for working with Kedro projects
    interactively in Jupyter notebooks or IPython.

    Parameters:
    - ipython: IPython instance

    Magic Commands Available:
    - %reload_kedro - Reload Kedro project
    - %load_node - Load a specific node for testing

    Usage:
    In Jupyter or IPython:
    >>> %load_ext kedro.ipython
    """

def reload_kedro(project_path: Path, env: str | None = None) -> dict[str, Any]:
    """
    Reload Kedro project context and catalog in IPython session.

    Parameters:
    - project_path: Path to Kedro project root
    - env: Environment name (e.g., 'local', 'prod')

    Returns:
    Dictionary with keys:
    - 'context': KedroContext instance
    - 'catalog': DataCatalog instance
    - 'session': KedroSession instance
    - 'pipelines': Dictionary of project pipelines

    Example:
    >>> from pathlib import Path
    >>> from kedro.ipython import reload_kedro
    >>> objects = reload_kedro(Path.cwd(), env='local')
    >>> catalog = objects['catalog']
    >>> context = objects['context']
    """

def magic_reload_kedro(line: str = "") -> dict[str, Any]:
    """
    IPython magic command to reload Kedro project.
    Accessible as %reload_kedro in IPython/Jupyter after loading extension.

    Parameters:
    - line: Optional arguments passed to magic command

    Returns:
    Dictionary with project objects (context, catalog, session, pipelines)

    Example (in IPython):
    >>> %reload_kedro
    >>> catalog.filter()  # Now you can use catalog
    """

def magic_load_node(line: str) -> str:
    """
    IPython magic command to generate code for loading and running a node.
    Accessible as %load_node in IPython/Jupyter after loading extension.

    Parameters:
    - line: Node name to load

    Returns:
    Generated Python code string for loading inputs and running the node

    Example (in IPython):
    >>> %load_node preprocessing_node
    # Generates and displays code like:
    # inputs = {
    #     'raw_data': catalog.load('raw_data')
    # }
    # outputs = node.run(inputs)
    """

# In Jupyter notebook
%load_ext kedro.ipython

# Reload project after code changes
%reload_kedro

# Access project context
context = %load_kedro_context
catalog = context.catalog
params = context.params

# Load extension
%load_ext kedro.ipython

# Access catalog
from kedro.framework.session import KedroSession

with KedroSession.create() as session:
    context = session.load_context()

    # Load data interactively
    data = context.catalog.load("input_data")

    # Process and explore
    print(f"Data shape: {data.shape}")
    print(data.head())

    # Test node functions
    from my_project.nodes import process_data
    result = process_data(data)

# In notebook with extension loaded
%load_ext kedro.ipython

# Test individual nodes
from my_project.pipelines.data_processing.nodes import clean_data

# Load test data
with KedroSession.create() as session:
    context = session.load_context()
    test_data = context.catalog.load("raw_data")

    # Test node
    cleaned = clean_data(test_data)
    print(f"Cleaned {len(cleaned)} records")

%load_ext kedro.ipython

# Run pipeline interactively
with KedroSession.create() as session:
    # Run specific nodes
    session.run(node_names=["clean_data", "create_features"])

    # Inspect outputs
    context = session.load_context()
    features = context.catalog.load("features")

    # Debug issues
    print(features.describe())