Special Purpose Classes and Decorators

class countable:
    def __init__(self, iterable: Iterable[Any]) -> None: ...
    def __iter__(self) -> Iterator[Any]: ...
    def items_seen(self) -> int: ...

from more_itertools import countable

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

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

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

# Continue consuming
more_items = [next(counter) for _ in range(2)]  # [3, 4]
counter.items_seen()  # 5

def consumer(func: Callable[..., Any]) -> Callable[..., Any]: ...
def make_decorator(wrapping_func: Callable[..., Any], *decorator_args: Any, **decorator_kwargs: Any) -> Callable[..., Any]: ...

from more_itertools import consumer

# Create a coroutine consumer function
@consumer
def printer():
    while True:
        item = yield
        print(f"Received: {item}")

# Use the consumer
p = printer()
p.send("Hello")     # Prints: Received: Hello  
p.send("World")     # Prints: Received: World

# Make custom decorators
from more_itertools import make_decorator

def logging_decorator(func, message):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}: {message}")
        return func(*args, **kwargs)
    return wrapper

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

@make_logger
def my_function(x):
    return x * 2

my_function(5)  # Prints: Calling my_function: Custom log message
                # Returns: 10

def raise_(exception: Exception) -> None: ...

from more_itertools import raise_

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

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

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

class AbortThread:
    def __init__(self) -> None: ...
    def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]: ...

from more_itertools import AbortThread
import threading
import time

# Create thread abort utility
abort = AbortThread()

@abort
def long_running_task():
    for i in range(100):
        if abort.should_abort():
            return f"Aborted at {i}"
        time.sleep(0.1)  # Simulate work
    return "Completed"

# Start thread
thread = threading.Thread(target=long_running_task)
thread.start()

# Abort after some time
time.sleep(0.5)
abort.set()

thread.join()