Iterator Utility Classes and Functions

0.2.0 · active · verified Sun Apr 12

The `iterators` library provides utility classes and functions for working with Python iterators. It includes features like adding timeouts to iterators and creating synchronous/asynchronous pipelines. The current version is 0.2.0, with an irregular release cadence as new utility features are added.

Warnings

gotcha As a pre-1.0.0 library (currently v0.2.0), this package does not strictly adhere to Semantic Versioning where minor versions guarantee backward compatibility. Breaking changes might occur between minor releases (e.g., 0.1.x to 0.2.x).
Fix: Always pin to exact versions (e.g., `pip install iterators==0.2.0`) or review release notes carefully before upgrading minor versions.
gotcha The `IteratorPipe` class, introduced in v0.2.0, is mentioned in release notes but lacks detailed examples or a dedicated section in the main GitHub README. Users might need to consult the source code or tests for advanced usage patterns, indicating a less stable API or incomplete documentation at this stage.
Fix: Refer to the library's test suite on GitHub for `IteratorPipe` examples, and exercise caution as its API might be subject to change in future minor releases.

Install

pip install iterators Install stable release

Imports

TimeoutIterator
```
from iterators import TimeoutIterator
```
IteratorPipe
```
from iterators import IteratorPipe
```

Quickstart

This example demonstrates the `TimeoutIterator` class, which wraps an existing iterator to introduce a timeout mechanism. If the next item isn't available within the specified timeout, a sentinel value is returned instead of waiting indefinitely or raising an exception. The timeout can also be adjusted dynamically. While `IteratorPipe` is a newer feature (v0.2.0), the most explicit quickstart examples in the official README focus on `TimeoutIterator`.

import time
from iterators import TimeoutIterator

def my_slow_iterator():
    yield 1
    time.sleep(0.6) # This sleep might cause a timeout
    yield 2
    time.sleep(0.4)
    yield 3

print("--- Demonstrating TimeoutIterator ---")
# Example 1: Basic iteration
i = my_slow_iterator()
it = TimeoutIterator(i)
print(f"Next: {next(it)}") # Expected: 1
print(f"Next: {next(it)}") # Expected: 2 (if timeout not hit)
print(f"Next: {next(it)}") # Expected: 3
try:
    next(it)
except StopIteration:
    print("StopIteration: Iterator exhausted.")

print("\n--- Demonstrating Timeout with Sentinel ---")
i_timeout = my_slow_iterator()
it_timeout = TimeoutIterator(i_timeout, timeout=0.5)
print(f"Next (with timeout): {next(it_timeout)}") # Expected: 1
# The next call will likely hit the timeout if sleep is 0.6s and timeout 0.5s
sentinel_value = next(it_timeout)
if sentinel_value == it_timeout.get_sentinel():
    print("Timeout occurred, sentinel value returned.")
else:
    print(f"Next (after potential timeout): {sentinel_value}")

# Continue to consume remaining items, potentially with adjusted timeout
it_timeout.set_timeout(0.7) # Adjust timeout dynamically
try:
    print(f"Next (after timeout reset): {next(it_timeout)}")
    print(f"Next (final item): {next(it_timeout)}")
except StopIteration:
    print("StopIteration: Iterator exhausted after timeout handling.")

view raw JSON →