Batch Consume Process Commit Pattern in Python

A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.

Medium Python 3.8+ Aug 9, 2026 Streaming & messaging 13 views 0 copies

streaming batch-processing queues consumer commit

Python code

53 lines

Python 3.8+

import random
import threading
import time
from collections import deque


class MockBatchProcessor:
    def __init__(self, process_func, commit_func, batch_size=5):
        self.queue = deque()
        self.batch_size = batch_size
        self.process_func = process_func
        self.commit_func = commit_func

    def consume(self, item):
        self.queue.append(item)
        if len(self.queue) >= self.batch_size:
            self._process_batch()

    def _process_batch(self):
        batch = [self.queue.popleft() for _ in range(self.batch_size)]
        try:
            results = self.process_func(batch)
            self.commit_func(batch, results, success=True)
        except Exception as exc:
            self.commit_func(batch, exc, success=False)

    def flush(self):
        while self.queue:
            single = [self.queue.popleft()]
            results = self.process_func(single)
            self.commit_func(single, results, success=True)


def process(items):
    time.sleep(0.01)  # Simulate work
    return [item * 2 for item in items]


def commit(items, results, success=True):
    status = "COMMITTED" if success else "FAILED"
    print(f"{status} batch {items} -> {results}")


if __name__ == "__main__":
    processor = MockBatchProcessor(process, commit, batch_size=3)

    for i in range(1, 8):
        processor.consume(i)
        if i == 4:
            # Simulate an error mid-stream
            processor.process_func = lambda x: (_ for _ in ()).throw(RuntimeError("mock failure"))

    processor.flush()

Output

stdout

COMMITTED batch [1, 2, 3] -> [2, 4, 6]
FAILED batch [4, 5, 6] -> mock failure
COMMITTED batch [7] -> [14]

How it works

This pattern separates consume, process, and commit phases — a common streaming architecture. The deque provides O(1) popleft operations for batch draining. The try/except inside _process_batch lets you commit failures explicitly, keeping the pipeline alive. The flush method handles leftovers when the stream ends, ensuring no items are lost. Batching amortizes per-item overhead and enables transactional-like commits.

Common mistakes

Forgetting to call flush(), which leaves unprocessed items in the queue
Processing and committing inside the same lock, killing throughput
Not handling partial batch failures — one bad item fails the entire batch
Using a list with pop(0) instead of deque, causing O(n) shifts

Variations

Use a bounded queue with maxlen to implement backpressure
Replace batch commit with per-item commits when latency matters more than throughput

Real-world use cases

Kafka consumers that aggregate messages into micro-batches before writing to a database.
ETL jobs that group incoming rows and commit them in transactions for atomicity.
Log aggregators that flush buffered events on a timer or when the buffer fills.

Batch Consume Process Commit Pattern in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes