processor.py

from typing import Any, Callable, Dict, List, TypeVar

T = TypeVar("T")
R = TypeVar("R")


class BatchProcessor:
    """Processes sequences of data in configurable batch sizes with custom transformations."""

    def __init__(
        self,
        batch_size: int = 100,
        skip_errors: bool = False,
    ) -> None:
        """Initialize the batch processor configuration.

        Args:
            batch_size: Maximum number of items per batch.
            skip_errors: Whether to ignore exceptions during item processing.
        """
        if batch_size <= 0:
            raise ValueError("batch_size must be greater than zero")
        self.batch_size = batch_size
        self.skip_errors = skip_errors

    def chunk(self, items: List[T]) -> List[List[T]]:
        """Split a list into smaller batches of size batch_size.

        Args:
            items: Sequence of items to divide into chunks.

        Returns:
            List of item batches.
        """
        return [
            items[i : i + self.batch_size]
            for i in range(0, len(items), self.batch_size)
        ]

    def process(
        self,
        items: List[T],
        func: Callable[[T], R],
    ) -> List[R]:
        """Apply a function to all items in batches.

        Args:
            items: Items to be processed.
            func: Transformation function applied to each item.

        Returns:
            List of successfully processed results.
        """
        results: List[R] = []
        for item in items:
            try:
                results.append(func(item))
            except Exception as err:
                if not self.skip_errors:
                    raise err
        return results


def process_dict_keys(
    data: Dict[str, Any],
    key_transform: Callable[[str], str],
) -> Dict[str, Any]:
    """Recursively transform dictionary keys using a provided callback.

    Args:
        data: Input dictionary to transform.
        key_transform: Function to modify each key string.

    Returns:
        New dictionary with transformed keys.
    """
    transformed: Dict[str, Any] = {}
    for key, value in data.items():
        new_key = key_transform(key)
        if isinstance(value, dict):
            transformed[new_key] = process_dict_keys(value, key_transform)
        elif isinstance(value, list):
            transformed[new_key] = [
                process_dict_keys(v, key_transform) if isinstance(v, dict) else v
                for v in value
            ]
        else:
            transformed[new_key] = value
    return transformed