0
# Utility Classes and Exceptions
1

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

4
## Capabilities
5

6
### Exception Classes
7

8
Custom exceptions for specific error conditions.
9

10
```python { .api }
11
class UnequalIterablesError(ValueError):
12
    def __init__(self, details: tuple[int, int, int] | None = None) -> None: ...
13
```
14

15
**Usage:**
16

17
```python
18
from more_itertools import UnequalIterablesError, zip_equal
19

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

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

34
# Usage
35
try:
36
    result = process_paired_data([1, 2, 3], [4, 5])
37
except UnequalIterablesError as e:
38
    print(f"Error: {e}")
39
```
40

41
### Type Safety and Error Handling
42

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

45
**Common Functions That May Raise This Error:**
46

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

51
**Best Practices:**
52

53
```python
54
from more_itertools import UnequalIterablesError
55

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

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

71
### Dynamic Grouping Classes
72

73
Classes for advanced grouping and bucketing operations.
74

75
```python { .api }
76
class bucket:
77
    """Dynamic bucketing of iterable items by key function."""
78
    
79
    def __init__(self, iterable, key, validator=None):
80
        """
81
        Initialize bucket grouping.
82
        
83
        Args:
84
            iterable: Source iterable to bucket
85
            key: Function to determine bucket assignment
86
            validator: Optional function to validate bucket keys
87
        """
88
    
89
    def __contains__(self, value):
90
        """Test if bucket key exists."""
91
    
92
    def __iter__(self):
93
        """Iterator over available bucket keys."""
94
    
95
    def __getitem__(self, value):
96
        """Get iterator for specific bucket."""
97
```
98

99
**Usage:**
100

101
```python
102
from more_itertools import bucket
103

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

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

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

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

121
### Callback Conversion Classes
122

123
Classes for converting callback-based functions to iterators.
124

125
```python { .api }
126
class callback_iter:
127
    """Convert callback-based function to iterator."""
128
    
129
    def __init__(self, func, callback_kwd='callback', wait_seconds=0.1):
130
        """
131
        Initialize callback iterator.
132
        
133
        Args:
134
            func: Function that accepts callback keyword argument
135
            callback_kwd: Name of callback parameter (default: 'callback')
136
            wait_seconds: Polling interval in seconds (default: 0.1)
137
        """
138
    
139
    def __enter__(self):
140
        """Enter context manager."""
141
    
142
    def __exit__(self, exc_type, exc_value, traceback):
143
        """Exit context manager."""
144
    
145
    def __iter__(self):
146
        """Return iterator object."""
147
    
148
    def __next__(self):
149
        """Return next callback invocation."""
150
    
151
    @property
152
    def done(self):
153
        """True if function execution completed."""
154
    
155
    @property
156
    def result(self):
157
        """Function result (only available after completion)."""
158
```
159

160
**Usage:**
161

162
```python
163
from more_itertools import callback_iter
164
import time
165

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

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

179
print(f"Final result: {it.result}")
180
# Output:
181
# Message: Step 1, Progress: 0.0%
182
# Message: Step 2, Progress: 25.0%
183
# Message: Step 3, Progress: 50.0%
184
# Message: Step 4, Progress: 75.0%
185
# Message: Step 5, Progress: 100.0%
186
# Final result: Complete
187
```