philiprehberger-batch-iter
v0.3.0352 downloadsMIT
Batch processing with progress tracking and error handling.
$
pip install philiprehberger-batch-iterREADME
philiprehberger-batch-iter
Batch processing with progress tracking and error handling.
Installation
pip install philiprehberger-batch-iter
Usage
from philiprehberger_batch_iter import batch, batch_map, collect_errors
# Split any iterable into fixed-size batches
for chunk in batch(range(10), size=3):
print(chunk)
# [0, 1, 2]
# [3, 4, 5]
# [6, 7, 8]
# [9]
# Enable progress output to stderr
for chunk in batch(range(100), size=25, progress=True):
process(chunk)
# batch 1: 25 items
# batch 2: 25 items
# ...
Batch map
from philiprehberger_batch_iter import batch_map
# Process items in batches and collect flattened results
results = batch_map(range(10), size=3, fn=lambda chunk: [x * 2 for x in chunk])
print(results)
# [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
Error collection
from philiprehberger_batch_iter import collect_errors
def process_batch(items):
for item in items:
if item < 0:
raise ValueError(f"negative value: {item}")
result = collect_errors([1, 2, -3, 4, 5, -6], size=2, fn=process_batch)
print(result.processed) # 6
print(len(result.errors)) # 2
print(result.duration_ms) # 0.12
Async batching
from philiprehberger_batch_iter import batch_async
async def process():
async for chunk in batch_async(async_data_source(), size=50):
await handle(chunk)
Async batch map
from philiprehberger_batch_iter import batch_async_map
async def upload(chunk):
return await api.upload_many(chunk)
results = await batch_async_map(items_aiter, size=100, fn=upload)
API
| Function / Class | Description |
|---|---|
batch(iterable, size, progress=False) | Yield fixed-size batches from an iterable |
batch_map(iterable, size, fn) | Process batches with fn and return a flat result list |
batch_async(async_iterable, size) | Async generator yielding fixed-size batches |
batch_async_map(async_iterable, size, fn) | Async counterpart to batch_map, awaits each batch |
collect_errors(iterable, size, fn) | Process batches and collect errors into a result |
BatchResult | Dataclass with processed, errors, duration_ms |
Development
pip install -e .
python -m pytest tests/ -v
Support
If you find this project useful: