Coverage for merco/hooks/registry.py: 89%

1"""钩子注册与调度""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import inspect 

7import logging 

8from dataclasses import dataclass 

9from typing import Callable 

10 

11logger = logging.getLogger("merco.hooks.registry") 

12 

13 

14@dataclass 

15class HookResult: 

16 """Hook handler 的结构化返回值。 

17 

18 data 合并进当前事件 kwargs，后续 handler 会看到更新后的 kwargs。 

19 stop=True 停止后续 hook handler；业务流程是否短路由调用方决定。 

20 """ 

21 

22 data: dict | None = None 

23 stop: bool = False 

24 

25 

26class HookRegistry: 

27 """事件钩子注册表""" 

28 

29 def __init__(self): 

30 self._hooks: dict[str, list[Callable]] = {} 

31 

32 def on(self, event: str, handler: Callable): 

33 """注册钩子处理器""" 

34 if event not in self._hooks: 

35 self._hooks[event] = [] 

36 self._hooks[event].append(handler) 

37 

38 def off(self, event: str, handler: Callable): 

39 """移除钩子处理器""" 

40 if event in self._hooks: 

41 self._hooks[event].remove(handler) 

42 

43 async def emit(self, event: str, **kwargs) -> HookResult | None: 

44 """触发事件。 

45 

46 默认 fire-and-forget：handler 返回 None 时不影响流程。 

47 handler 可返回 HookResult(data=...) 修改后续 handler 看到的 kwargs。 

48 handler 可返回 HookResult(stop=True) 停止后续 handler 链。 

49 """ 

50 handlers = self._hooks.get(event, []) 

51 current = dict(kwargs) 

52 changed = False 

53 

54 for handler in handlers: 

55 try: 

56 result = handler(**current) 

57 if inspect.isawaitable(result): 

58 result = await result 

59 except Exception: 

60 logger.debug("hook %s handler error", event, exc_info=True) 

61 continue 

62 

63 if isinstance(result, HookResult): 

64 if result.data: 

65 current.update(result.data) 

66 changed = True 

67 if result.stop: 

68 return HookResult(data=current, stop=True) 

69 

70 if changed: 

71 return HookResult(data=current, stop=False) 

72 return None 

73 

74 def clear(self, event: str = None): 

75 """清除钩子""" 

76 if event: 

77 self._hooks.pop(event, None) 

78 else: 

79 self._hooks.clear()