Framework Utilities API Reference

class CatalogCommandsMixin:
    """
    Mixin providing catalog-related command methods for context.
    Adds catalog inspection and pattern resolution capabilities.
    """

    def describe_datasets(
        self,
        pipelines: dict[str, Pipeline] | None = None,
    ) -> dict[str, dict[str, Any]]:
        """
        Generate detailed descriptions for all datasets in the catalog.

        Parameters:
        - pipelines: Optional dictionary of pipeline names to Pipeline objects.
                    Used to determine which datasets are used in pipelines.
                    Defaults to None

        Returns:
        Dictionary mapping dataset names to their descriptions, including:
        - type: Dataset class name
        - filepath: File path (if applicable)
        - layer: Data layer (if using DataCatalog layers)
        - used_in_pipelines: List of pipeline names using this dataset
        """

    def list_patterns(self) -> list[str]:
        """
        List all dataset factory patterns registered in the catalog.

        Returns:
        List of pattern strings (e.g., '{default}#csv', 'params:{name}')
        """

    def resolve_patterns(
        self,
        pipelines: dict[str, Pipeline] | None = None,
    ) -> dict[str, dict[str, Any]]:
        """
        Resolve dataset factory patterns and return information about resolved datasets.

        Parameters:
        - pipelines: Optional dictionary of pipeline names to Pipeline objects.
                    Used to determine which patterns are used in pipelines.
                    Defaults to None

        Returns:
        Dictionary mapping resolved dataset names to their configuration details
        """

def compose_classes(*classes: type) -> type:
    """
    Compose multiple classes into a single class using multiple inheritance.

    Useful for combining mixins with base context classes.

    Parameters:
    - classes: Variable number of classes to compose

    Returns:
    New composed class type

    Example:
    >>> CustomContext = compose_classes(KedroContext, CatalogCommandsMixin)
    """

from kedro.framework.context import KedroContext
from kedro.framework.context.utilities import CatalogCommandsMixin, compose_classes

# Create custom context with catalog commands
CustomContext = compose_classes(KedroContext, CatalogCommandsMixin)

# Use in session
with KedroSession.create() as session:
    context = session.load_context()
    datasets = context.list_datasets(patterns=["^params:"])
    print(f"Parameters: {datasets}")

class CustomMixin:
    """Custom functionality mixin."""
    def custom_method(self):
        return "custom"

# Compose multiple mixins
EnhancedContext = compose_classes(
    KedroContext,
    CatalogCommandsMixin,
    CustomMixin
)

from kedro.framework.context import KedroContext
from kedro.framework.context.utilities import compose_classes

class MonitoringMixin:
    """Add monitoring capabilities to context."""

    def track_dataset_load(self, dataset_name: str):
        """Track when datasets are loaded."""
        print(f"Loading dataset: {dataset_name}")
        # Could send metrics to monitoring service

    def track_pipeline_execution(self, pipeline_name: str):
        """Track pipeline executions."""
        print(f"Executing pipeline: {pipeline_name}")
        # Could log to monitoring dashboard


class ValidationMixin:
    """Add data validation capabilities."""

    def validate_catalog(self):
        """Validate all datasets in catalog."""
        datasets = self.catalog.list()
        for ds_name in datasets:
            if not self.catalog.exists(ds_name):
                print(f"Warning: {ds_name} doesn't exist")


# Compose all mixins into custom context
MonitoredContext = compose_classes(
    KedroContext,
    CatalogCommandsMixin,
    MonitoringMixin,
    ValidationMixin
)

# Configure in settings.py
# CONTEXT_CLASS = MonitoredContext

from kedro.framework.session import KedroSession

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

    # List all dataset patterns
    patterns = context.list_patterns()
    print(f"Dataset patterns: {patterns}")

    # Describe all datasets
    datasets_info = context.describe_datasets()
    for name, info in datasets_info.items():
        print(f"{name}: {info['type']}")

    # Resolve pattern-based datasets
    resolved = context.resolve_patterns()
    print(f"Resolved datasets: {list(resolved.keys())}")

# Order matters - mixins should come after base class
GoodContext = compose_classes(KedroContext, MixinA, MixinB)

# MRO: GoodContext -> KedroContext -> MixinA -> MixinB -> object

# If mixins have conflicting methods, earlier classes take precedence:
class MixinA:
    def method(self): return "A"

class MixinB:
    def method(self): return "B"

Context = compose_classes(KedroContext, MixinA, MixinB)
# Context().method() returns "A" (MixinA comes first)

# Mixin approach (composable, reusable)
Context = compose_classes(KedroContext, FeatureA, FeatureB)

# Inheritance approach (single purpose)
class ProjectContext(KedroContext):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.project_config = self._load_project_config()