หาบั๊กให้เจอ: Data Engineering

1from typing import Callable
2
3class PipelineEventBus:
4    def __init__(self):
5        self._subscribers: list[Callable] = []
6
7    def subscribe(self, callback: Callable):
8        self._subscribers.append(callback)
9
10    def publish(self, event: dict):
11        for sub in self._subscribers:
12            sub(event)
13
14bus = PipelineEventBus()
15bus.subscribe(lambda e: metrics.record(e))
16bus.subscribe(lambda e: alerting.notify(e))
17bus.subscribe(lambda e: audit_log.write(e))
18
19def on_batch_complete(batch_id: str):
20    bus.publish({"type": "batch_complete", "batch_id": batch_id})