Coverage for src / kemi / operations / _ops_webhooks.py: 26%

1"""Webhook operations: configure_webhooks, _dispatch_webhook_event. 

2 

3These free functions are called by the corresponding ``Memory`` methods. 

4""" 

5 

6from __future__ import annotations 

7 

8import asyncio 

9import logging 

10from typing import TYPE_CHECKING, Any 

11 

12from kemi.webhooks import WebhookDispatcher, WebhookEventType, WebhookStore, build_payload 

13 

14if TYPE_CHECKING: 

15 from kemi._memory_impl import Memory 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20def configure(memory: "Memory", db_path: str | None) -> None: 

21 """Enable webhook dispatch for memory lifecycle events.""" 

22 if db_path is None: 

23 try: 

24 db_path = memory._store._db_path # type: ignore[attr-defined] 

25 except AttributeError: 

26 logger.warning("Cannot determine database path for webhook store") 

27 return 

28 

29 try: 

30 store = WebhookStore(db_path=db_path) 

31 memory._webhook_dispatcher = WebhookDispatcher(store=store) 

32 logger.info("Webhook dispatcher initialized (db: %s)", db_path) 

33 except (OSError, ValueError) as e: 

34 logger.warning("Failed to initialise webhook dispatcher: %s", e) 

35 

36 

37def dispatch( 

38 memory: "Memory", 

39 event: WebhookEventType, 

40 memory_id: str, 

41 user_id: str, 

42 snapshot: dict[str, Any] | None = None, 

43 previous_state: dict[str, Any] | None = None, 

44 **extra: Any, 

45) -> None: 

46 """Dispatch a webhook event if a dispatcher is configured. 

47 

48 Prefers async dispatch when an event loop is running; falls back to 

49 synchronous dispatch otherwise (e.g. CLI commands). 

50 """ 

51 if memory._webhook_dispatcher is None: 

52 return 

53 try: 

54 payload = build_payload( 

55 event=event, 

56 memory_id=memory_id, 

57 user_id=user_id, 

58 snapshot=snapshot, 

59 previous_state=previous_state, 

60 extra=extra or None, 

61 ) 

62 except (ValueError, TypeError) as e: 

63 logger.warning("Webhook payload build failed for %s: %s", event.value, e) 

64 return 

65 

66 try: 

67 loop = asyncio.get_running_loop() 

68 except RuntimeError: 

69 # No event loop — fall back to sync dispatch. 

70 try: 

71 memory._webhook_dispatcher.dispatch_sync(payload, event) 

72 except Exception: 

73 # Broad catch: webhook transports vary (HTTP, CLI subprocess, etc.) 

74 # Log and continue; webhooks must never break the calling operation. 

75 logger.warning( 

76 "Sync webhook dispatch failed for %s", event.value, exc_info=True 

77 ) 

78 return 

79 

80 # Running event loop — fire-and-forget. 

81 try: 

82 asyncio.ensure_future( 

83 memory._webhook_dispatcher.dispatch_async(payload, event) 

84 ) 

85 except Exception: 

86 # Broad catch: ensure_future can raise if loop is closing, etc. 

87 logger.warning( 

88 "Async webhook dispatch failed for %s", event.value, exc_info=True 

89 )