How to Simulate a Micro-Batch Interval Trigger in Python

import time
from dataclasses import dataclass, field
from typing import List, Callable


@dataclass
class MicroBatchTriggerMock:
    batch_interval_seconds: float = 0.5
    max_batches: int = 5
    _batches_emitted: int = 0
    _next_emit_time: float = field(init=False, default=0)

    def start(self, on_batch: Callable[[int], None]):
        self._next_emit_time = time.time() + self.batch_interval_seconds
        while self._batches_emitted < self.max_batches:
            now = time.time()
            if now >= self._next_emit_time:
                self._batches_emitted += 1
                on_batch(self._batches_emitted)
                self._next_emit_time += self.batch_interval_seconds
            else:
                time.sleep(0.05)


def example_consumer(batch_number: int):
    print(f"Batch {batch_number} processed at {time.time():.2f}")


if __name__ == "__main__":
    trigger = MicroBatchTriggerMock(batch_interval_seconds=0.3, max_batches=3)
    trigger.start(example_consumer)
    print(f"Finished: {trigger._batches_emitted} batches emitted")