How to Track Checkpoint Offset After Batch Commit in Python

import json
from typing import Any


class BatchProcessor:
    """Tracks checkpoint offset after committing batches."""

    def __init__(self, batch_size: int = 3):
        self.batch_size = batch_size
        self.offset = 0  # last successfully committed offset (exclusive)
        self.total_committed = 0

    def process_and_commit(self, records: list[Any]) -> int:
        """Process records in batches, advancing checkpoint only on full commit."""
        for i in range(0, len(records), self.batch_size):
            batch = records[i:i + self.batch_size]
            # Simulate successful commit
            self.total_committed += len(batch)
            self.offset = i + len(batch)  # checkpoint = end of this batch
        return self.offset

    def get_state(self) -> dict:
        return {
            "last_checkpoint_offset": self.offset,
            "total_committed": self.total_committed,
            "pending": False,
        }


if __name__ == "__main__":
    processor = BatchProcessor(batch_size=3)
    events = ["evt_1", "evt_2", "evt_3", "evt_4", "evt_5", "evt_6", "evt_7"]
    final_offset = processor.process_and_commit(events)
    print(json.dumps(processor.get_state(), indent=2))
    print(f"Final checkpoint offset: {final_offset}")