tessl/pypi-more-itertools

Additional building blocks, recipes, and routines for working with Python iterables beyond itertools.

—

Pending

Overview

Eval results

Files

Utility Classes and Exceptions

Name: tessl/pypi-more-itertools
Author: tessl

Exception classes and utility types for error handling and type safety.

Capabilities

Exception Classes

Custom exceptions for specific error conditions.

class UnequalIterablesError(ValueError):
    def __init__(self, details: tuple[int, int, int] | None = None) -> None: ...

Usage:

from more_itertools import UnequalIterablesError, zip_equal

# This exception is raised by functions that expect equal-length iterables
try:
    # zip_equal raises UnequalIterablesError if lengths differ
    result = list(zip_equal([1, 2, 3], [4, 5]))  
except UnequalIterablesError as e:
    print(f"Iterables have different lengths: {e}")

# You can also raise it in your own code for consistency
def process_paired_data(list1, list2):
    if len(list1) != len(list2):
        raise UnequalIterablesError(f"Expected equal lengths, got {len(list1)} and {len(list2)}")
    
    return [a + b for a, b in zip(list1, list2)]

# Usage
try:
    result = process_paired_data([1, 2, 3], [4, 5])
except UnequalIterablesError as e:
    print(f"Error: {e}")

Type Safety and Error Handling

The UnequalIterablesError provides better error messages and type safety for operations that require equal-length iterables, making debugging easier and code more robust.

Common Functions That May Raise This Error:

zip_equal() - Ensures iterables have same length when zipping
sort_together() - Requires all sequences to have same length
Other functions that assume parallel iteration over equal-length sequences

Best Practices:

from more_itertools import UnequalIterablesError

def safe_parallel_operation(iter1, iter2):
    """Example of defensive programming with UnequalIterablesError"""
    try:
        # Attempt operation that requires equal lengths
        from more_itertools import zip_equal
        return list(zip_equal(iter1, iter2))
    except UnequalIterablesError:
        # Handle gracefully or provide fallback
        print("Warning: Iterables have different lengths, truncating to shorter")
        return list(zip(iter1, iter2))

# This provides clear error messages and allows for graceful handling
result = safe_parallel_operation([1, 2, 3, 4], [5, 6])

Dynamic Grouping Classes

Classes for advanced grouping and bucketing operations.

class bucket:
    """Dynamic bucketing of iterable items by key function."""
    
    def __init__(self, iterable, key, validator=None):
        """
        Initialize bucket grouping.
        
        Args:
            iterable: Source iterable to bucket
            key: Function to determine bucket assignment
            validator: Optional function to validate bucket keys
        """
    
    def __contains__(self, value):
        """Test if bucket key exists."""
    
    def __iter__(self):
        """Iterator over available bucket keys."""
    
    def __getitem__(self, value):
        """Get iterator for specific bucket."""

Usage:

from more_itertools import bucket

# Group students by grade level
students = ['A1', 'B1', 'A2', 'C1', 'B2', 'A3']
by_grade = bucket(students, key=lambda x: x[0])

# Access specific buckets
a_students = list(by_grade['A'])  # ['A1', 'A2', 'A3']
b_students = list(by_grade['B'])  # ['B1', 'B2']

# Check available buckets
available_grades = sorted(list(by_grade))  # ['A', 'B', 'C']

# With validator for infinite iterables
from itertools import count, islice
numbers = bucket(count(1), key=lambda x: x % 3, validator=lambda x: x in {0, 1, 2})
mod0_numbers = list(islice(numbers[0], 5))  # [3, 6, 9, 12, 15]

Callback Conversion Classes

Classes for converting callback-based functions to iterators.

class callback_iter:
    """Convert callback-based function to iterator."""
    
    def __init__(self, func, callback_kwd='callback', wait_seconds=0.1):
        """
        Initialize callback iterator.
        
        Args:
            func: Function that accepts callback keyword argument
            callback_kwd: Name of callback parameter (default: 'callback')
            wait_seconds: Polling interval in seconds (default: 0.1)
        """
    
    def __enter__(self):
        """Enter context manager."""
    
    def __exit__(self, exc_type, exc_value, traceback):
        """Exit context manager."""
    
    def __iter__(self):
        """Return iterator object."""
    
    def __next__(self):
        """Return next callback invocation."""
    
    @property
    def done(self):
        """True if function execution completed."""
    
    @property
    def result(self):
        """Function result (only available after completion)."""

Usage:

from more_itertools import callback_iter
import time

def progress_function(callback=None):
    """Function that reports progress via callback"""
    for i in range(5):
        time.sleep(0.1)  # Simulate work
        if callback:
            callback(f"Step {i+1}", progress=i/4)
    return "Complete"

# Convert callback-based function to iterator
with callback_iter(progress_function) as it:
    for args, kwargs in it:
        print(f"Message: {args[0]}, Progress: {kwargs['progress']:.1%}")

print(f"Final result: {it.result}")
# Output:
# Message: Step 1, Progress: 0.0%
# Message: Step 2, Progress: 25.0%
# Message: Step 3, Progress: 50.0%
# Message: Step 4, Progress: 75.0%
# Message: Step 5, Progress: 100.0%
# Final result: Complete

Install with Tessl CLI