Coverage for polar/kit/hook.py: 47%

1from collections.abc import Callable, Coroutine 

2from typing import Any 

3 

4type HookFunc[T] = Callable[[T], Coroutine[Any, Any, Any]] 

5 

6 

7class Hook[T]: 

8 hooks: list[HookFunc[T]] 

9 

10 def __init__(self) -> None: 

11 self.hooks = [] 

12 

13 def add(self, fun: HookFunc[T]) -> None: 

14 if fun in self.hooks: 

15 raise Exception(f"fun is already registered! fun={fun} hooks={self.hooks}") 

16 

17 self.hooks.append(fun) 

18 

19 async def call(self, payload: T) -> None: 

20 for fn in self.hooks: 

21 await fn(payload)