Python EventBus

問題
#

之前在找 python 的 EventBus 相關 lib，才發現寫的爛的星數高，寫的好的星數低。
以我翻到的兩個星數稍高的 lib 為例：

blinker: 2k stars，據說是 flask 用的 signal 系統。
pyee: 4xx stars，好像是抄 node 的。

可以看到這兩個 lib 有個共通特點，就是用字串當 signal，參數為不定參數。

雖然不知道 python 用戶到底多癡迷於字串，但 EventBus 通常是要處理兩個部份。其一為發生的事件，其二為事件的夾帶訊息。
使用字串+不定參數根本無法綁定這兩者的關係。以 blinker 提供的範例程式碼為例：

from blinker import signal

started = signal('round-started')
def each(round):
    print(f"Round {round}")

started.connect(each)

for round in range(1, 4):
    started.send(round)

each function 預設了一定會有一個 round 參數，但如果 started.send 少傳遞此資訊、或是多傳遞一個參數，根本不會有任何警告，只能等待程式在 runtime 爆掉。
「小心一點」可以解決這個問題，只是要用肉眼檢查原本 linter 可以幫忙檢查的東西，實在是蠢到不能在蠢了。

儘管九成 lib 都是採用字串低能設計，但還是有像 bubus 這種作法正確的 package 存在。可惜目前的星數是驚人的 87 顆。在 python 社群做事可能就是要習慣低能設計被人瘋狂追捧，正確設計無人問津的狀況。

練習
#

簡單系統
#

其實自己訂製一個 event bus 也不是很難，有趣的是如何讓使用的人不會出錯。
以下是一個簡單的 event bus 系統。

from typing import Any, Callable, NamedTuple, TypeVar

T = TypeVar("T")
_listener: dict[type, list[Callable[[Any], Any]]] = {}


def on(event_type: type[T], listener: Callable[[T], Any]):
    if event_type not in _listener:
        _listener[event_type] = []
    _listener[event_type].append(listener)


def emit(event):
    event_type = type(event)
    if event_type in _listener:
        for listener in _listener[event_type]:
            listener(event)

需要注意的只有 on 的定義。on 接受兩個參數：
參數1: 一個型別，代表要監聽的 event 種類。
參數2: 一個 function，且此 function 的第一個參數為剛剛監聽的型別類型。

使用時

class MyEvent(NamedTuple):
    message: str

def on_event(e: MyEvent) -> None:
    print(f"Received event with message: {e.message}")

on(MyEvent, on_event)

emit(MyEvent(message="Hello, World!"))

如果今天使用者想監聽不正確的 event，例如：

def on_event_2(a: str):
    print("hello", a)

on(MyEvent, on_event_2) # 紅線: Argument of type "(a: str) -> None" cannot be assigned to parameter "listener" of type "(T@on) -> Any" in function "on"

就會有紅線標注說你聽錯了

decorator
#

可以更進一步把這個東西化成 decorator，由於註冊時需要監聽的型別，很容易就會寫成下面這樣

@on_deco(MyEvent)
def on_event(e: MyEvent):
    pass

但仔細想想，function parameter 其實已經標注了監聽型別了，實在沒必要把同樣的東西寫兩次。因此更為理想的 decorator 為

@on_deco
def on_event(e: MyEvent):
    pass

實作此 decorator 需要使用到一點 python 魔術。

def on_deco(func: Callable[[T], Any]) -> Callable[[T], Any]:
    # 1. 取得 function signature
    sig = inspect.signature(func)

    # 2. 取得所有參數
    params = list(sig.parameters.values())

    # 3. 取出第一個參數的 annotation
    if len(params) == 1:
        on(params[0].annotation, func)
    else:
        raise RuntimeError("Decorator can only be used on functions with one parameter")
    return func

如此一來就能在不重複宣告監聽類別的情況下，使用 decorator 進行註冊。

問題#

練習#

簡單系統#

decorator#

問題
#

練習
#

簡單系統
#

decorator
#