Coverage for astrocyte/_hooks.py: 100%

1"""Hook manager — registration and dispatch for event hooks.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import logging 

7import threading 

8import uuid 

9from collections.abc import Awaitable, Callable 

10from datetime import datetime, timezone 

11 

12from astrocyte.policy.observability import StructuredLogger 

13from astrocyte.types import HookEvent 

14 

15# Hook handler type — FFI-safe: takes a HookEvent, returns an awaitable or None. 

16HookHandler = Callable[[HookEvent], Awaitable[None] | None] 

17 

18 

19class HookManager: 

20 """Thread-safe event hook registration and dispatch.""" 

21 

22 def __init__(self, logger: StructuredLogger) -> None: 

23 self._hooks: dict[str, list[HookHandler]] = {} 

24 self._lock = threading.Lock() 

25 self._logger = logger 

26 

27 def register(self, event_type: str, handler: HookHandler) -> None: 

28 """Register an event hook handler.""" 

29 with self._lock: 

30 if event_type not in self._hooks: 

31 self._hooks[event_type] = [] 

32 self._hooks[event_type].append(handler) 

33 

34 async def fire( 

35 self, 

36 event_type: str, 

37 bank_id: str | None = None, 

38 data: dict[str, str | int | float | bool | None] | None = None, 

39 ) -> None: 

40 """Fire all registered hooks for an event type. Non-blocking, failures logged.""" 

41 with self._lock: 

42 handlers = list(self._hooks.get(event_type, [])) 

43 if not handlers: 

44 return 

45 event = HookEvent( 

46 event_id=uuid.uuid4().hex, 

47 type=event_type, 

48 timestamp=datetime.now(timezone.utc), 

49 bank_id=bank_id, 

50 data=data, 

51 ) 

52 for handler in handlers: 

53 try: 

54 result = handler(event) 

55 if asyncio.iscoroutine(result) or asyncio.isfuture(result): 

56 await result 

57 except Exception: 

58 self._logger.log( 

59 "astrocyte.hook.error", 

60 bank_id=bank_id, 

61 data={"event_type": event_type}, 

62 level=logging.WARNING, 

63 )