philiprehberger-batch-iter 0.3.0

Batch processing with progress tracking and error handling.

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:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT