Coverage for src / lexigram / contracts / core / hooks.py: 0%

1"""Hook registry protocol and priority enum for framework extensibility.""" 

2 

3from __future__ import annotations 

4 

5from enum import IntEnum 

6from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

7 

8if TYPE_CHECKING: 

9 from collections.abc import Callable 

10 

11 

12class HookPriority(IntEnum): 

13 """Named priority levels for hook handlers. 

14 

15 Lower values run first. Custom integer values are accepted anywhere 

16 a priority is expected — these named levels are conveniences. 

17 

18 Spacing between levels leaves room for application-specific 

19 intermediate values (e.g. ``HookPriority.EARLY + 10``). 

20 """ 

21 

22 EARLIEST = 0 

23 EARLY = 50 

24 NORMAL = 100 

25 LATE = 200 

26 LATEST = 300 

27 

28 

29@runtime_checkable 

30class HookRegistryProtocol(Protocol): 

31 """Structural protocol for hook registries. 

32 

33 Defines the contract for action hooks (fire-and-forget, error-isolated) 

34 and filter hooks (value-pipeline, error-propagating) with priority ordering. 

35 """ 

36 

37 def register_action( 

38 self, 

39 hook_name: str, 

40 handler: Callable[..., Any], 

41 priority: int = 100, 

42 *, 

43 once: bool = False, 

44 ) -> None: 

45 """Register an action handler for the given hook. 

46 

47 Args: 

48 hook_name: The hook point name. 

49 handler: Callable to invoke. Can be sync or async. 

50 priority: Lower values run first. Defaults to 100. 

51 once: If True, auto-remove handler after first invocation. 

52 """ 

53 ... 

54 

55 def register_filter( 

56 self, 

57 hook_name: str, 

58 handler: Callable[..., Any], 

59 priority: int = 100, 

60 *, 

61 once: bool = False, 

62 ) -> None: 

63 """Register a filter handler for the given hook. 

64 

65 Args: 

66 hook_name: The hook point name. 

67 handler: Callable that receives and returns a value. 

68 priority: Lower values run first. Defaults to 100. 

69 once: If True, auto-remove handler after first invocation. 

70 """ 

71 ... 

72 

73 def unregister_action(self, hook_name: str, handler: Callable[..., Any]) -> bool: 

74 """Remove an action handler. Returns True if found and removed.""" 

75 ... 

76 

77 def unregister_filter(self, hook_name: str, handler: Callable[..., Any]) -> bool: 

78 """Remove a filter handler. Returns True if found and removed.""" 

79 ... 

80 

81 async def call_action(self, hook_name: str, **kwargs: Any) -> None: 

82 """Invoke all action handlers. Errors are isolated and logged.""" 

83 ... 

84 

85 async def apply_filter(self, hook_name: str, value: Any, **kwargs: Any) -> Any: 

86 """Pass value through all filter handlers. Errors propagate.""" 

87 ... 

88 

89 def has_action(self, hook_name: str) -> bool: 

90 """Check if any action handlers are registered.""" 

91 ... 

92 

93 def has_filter(self, hook_name: str) -> bool: 

94 """Check if any filter handlers are registered.""" 

95 ... 

96 

97 def clear(self, hook_name: str | None = None) -> None: 

98 """Clear handlers for a specific hook, or all hooks.""" 

99 ... 

100 

101 

102__all__ = [ 

103 "HookPriority", 

104 "HookRegistryProtocol", 

105]