0
# Special Purpose Classes and Decorators
1

2
Classes and decorators providing specialized functionality for advanced use cases.
3

4
## Capabilities
5

6
### Iterator Wrappers
7

8
Classes that provide additional functionality when wrapping iterators.
9

10
```python { .api }
11
class countable:
12
    def __init__(self, iterable: Iterable[Any]) -> None: ...
13
    def __iter__(self) -> Iterator[Any]: ...
14
    def items_seen(self) -> int: ...
15
```
16

17
**Usage:**
18

19
```python
20
from more_itertools import countable
21

22
# Track how many items have been consumed
23
data = range(10)
24
counter = countable(data)
25

26
# Consume some items
27
items = [next(counter) for _ in range(3)]  # [0, 1, 2]
28

29
# Check how many items have been seen
30
counter.items_seen()  # 3
31

32
# Continue consuming
33
more_items = [next(counter) for _ in range(2)]  # [3, 4]
34
counter.items_seen()  # 5
35
```
36

37
### Function Decorators
38

39
Decorators for creating specialized function behaviors.
40

41
```python { .api }
42
def consumer(func: Callable[..., Any]) -> Callable[..., Any]: ...
43
def make_decorator(wrapping_func: Callable[..., Any], *decorator_args: Any, **decorator_kwargs: Any) -> Callable[..., Any]: ...
44
```
45

46
**Usage:**
47

48
```python
49
from more_itertools import consumer
50

51
# Create a coroutine consumer function
52
@consumer
53
def printer():
54
    while True:
55
        item = yield
56
        print(f"Received: {item}")
57

58
# Use the consumer
59
p = printer()
60
p.send("Hello")     # Prints: Received: Hello  
61
p.send("World")     # Prints: Received: World
62

63
# Make custom decorators
64
from more_itertools import make_decorator
65

66
def logging_decorator(func, message):
67
    def wrapper(*args, **kwargs):
68
        print(f"Calling {func.__name__}: {message}")
69
        return func(*args, **kwargs)
70
    return wrapper
71

72
# Create a decorator factory
73
make_logger = make_decorator(logging_decorator, "Custom log message")
74

75
@make_logger
76
def my_function(x):
77
    return x * 2
78

79
my_function(5)  # Prints: Calling my_function: Custom log message
80
                # Returns: 10
81
```
82

83
### Utility Functions
84

85
Utility functions for specialized operations.
86

87
```python { .api }
88
def raise_(exception: Exception) -> None: ...
89
```
90

91
**Usage:**
92

93
```python
94
from more_itertools import raise_
95

96
# Utility for raising exceptions in expressions (useful in lambdas)
97
data = [1, 2, -1, 4, 5]
98

99
# Use in conditional expressions
100
positive_or_error = [x if x > 0 else raise_(ValueError("Negative value")) for x in data if x != -1]
101
# Will raise ValueError when encountering -1
102

103
# More practical use in error handling
104
def safe_divide(a, b):
105
    return a / b if b != 0 else raise_(ZeroDivisionError("Division by zero"))
106
```
107

108
### Threading Utilities
109

110
Utilities for threading operations.
111

112
```python { .api }
113
class AbortThread:
114
    def __init__(self) -> None: ...
115
    def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]: ...
116
```
117

118
**Usage:**
119

120
```python
121
from more_itertools import AbortThread
122
import threading
123
import time
124

125
# Create thread abort utility
126
abort = AbortThread()
127

128
@abort
129
def long_running_task():
130
    for i in range(100):
131
        if abort.should_abort():
132
            return f"Aborted at {i}"
133
        time.sleep(0.1)  # Simulate work
134
    return "Completed"
135

136
# Start thread
137
thread = threading.Thread(target=long_running_task)
138
thread.start()
139

140
# Abort after some time
141
time.sleep(0.5)
142
abort.set()
143

144
thread.join()
145
```