How to Build a Materialized View Updater Consumer Mock in Python

import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional


@dataclass
class MaterializedViewUpdater:
    """Mock updater that consumes change events and refreshes a view."""
    refresh: Optional[Callable[[str], None]] = None
    queue: Deque[tuple] = field(default_factory=deque)

    def consume(self, event: tuple) -> None:
        """Accept a change event, add to queue, and attempt refresh."""
        self.queue.append(event)
        if self.refresh:
            self.refresh(event[1])

    def drain(self) -> int:
        """Process all queued events with simple simulated work."""
        count = 0
        while self.queue:
            _, entity_id = self.queue.popleft()
            time.sleep(0.01)  # Simulate refresh latency
            count += 1
        return count


def example_refresh(entity_id: str) -> None:
    print(f"Refreshing view for {entity_id}")


if __name__ == "__main__":
    updater = MaterializedViewUpdater(refresh=example_refresh)
    events = [("INSERT", "user_42"), ("UPDATE", "order_7"), ("DELETE", "order_7")]
    for ev in events:
        updater.consume(ev)
    print(f"Processed {updater.drain()} events")