codekingpro/portable-devtools

1from __future__ import annotations2 3import logging4from collections.abc import Callable5from threading import Lock6from typing import TypedDict7 8from typing_extensions import NotRequired9 10logger = logging.getLogger(__name__)11 12 13class SerdeEvent(TypedDict):14    kind: str15    module: str16    name: str17    method: NotRequired[str]18 19 20SerdeEventListener = Callable[[SerdeEvent], None]21 22_listeners: list[SerdeEventListener] = []23_listeners_lock = Lock()24 25 26def register_serde_event_listener(listener: SerdeEventListener) -> Callable[[], None]:27    """Register a listener for serde allowlist events."""28    with _listeners_lock:29        _listeners.append(listener)30 31    def unregister() -> None:32        with _listeners_lock:33            try:34                _listeners.remove(listener)35            except ValueError:36                pass37 38    return unregister39 40 41def emit_serde_event(event: SerdeEvent) -> None:42    """Emit a serde event to all listeners.43 44    Listener failures are isolated and logged.45    """46    with _listeners_lock:47        listeners = tuple(_listeners)48    for listener in listeners:49        try:50            listener(event)51        except Exception:52            logger.warning("Serde listener failed", exc_info=True)53