Coverage for src/lumen/cortex/registry.py: 99%

1""" 

2The process-wide list of registered plugins. 

3 

4Registration is the whole contract for getting code into the server: a plugin 

5is named in configuration (or handed to :func:`lumen.register_plugin` in a 

6development session) and everything it declared is wired up here. Adding a 

7plugin never touches the core. 

8""" 

9 

10from __future__ import annotations 

11 

12from collections.abc import Awaitable, Callable 

13from dataclasses import dataclass 

14from typing import Any 

15 

16import structlog 

17 

18from .contract import PluginRegistrationError, check_version 

19from .plugin import HookSpec, OperationSpec, Plugin 

20 

21logger = structlog.get_logger(__name__) 

22 

23 

24@dataclass(frozen=True) 

25class RegisteredHook: 

26 plugin_name: str 

27 plugin_version: str 

28 spec: HookSpec 

29 callback: Callable[..., Awaitable[Any]] 

30 

31 

32@dataclass(frozen=True) 

33class RegisteredOperation: 

34 plugin_name: str 

35 plugin_version: str 

36 spec: OperationSpec 

37 callback: Callable[..., Awaitable[Any]] 

38 

39 

40class PluginRegistry: 

41 """Which plugins are loaded and what they listen to.""" 

42 

43 def __init__(self) -> None: 

44 self._plugins: dict[str, Plugin] = {} 

45 self._hooks: list[RegisteredHook] = [] 

46 self._operations: list[RegisteredOperation] = [] 

47 

48 # ── State ───────────────────────────────────────────────────────────────── 

49 

50 @property 

51 def plugins(self) -> list[Plugin]: 

52 return list(self._plugins.values()) 

53 

54 @property 

55 def has_hooks(self) -> bool: 

56 """False in the default configuration — the write path checks this first.""" 

57 return bool(self._hooks) 

58 

59 @property 

60 def operations(self) -> list[RegisteredOperation]: 

61 return list(self._operations) 

62 

63 def hooks_for(self, event: str, resource_type: str) -> list[RegisteredHook]: 

64 return [h for h in self._hooks if h.spec.matches(event, resource_type)] 

65 

66 # ── Registration ────────────────────────────────────────────────────────── 

67 

68 def register(self, plugin: Plugin, *, replace: bool = False) -> None: 

69 """Wire up ``plugin``'s hooks and operations. 

70 

71 ``replace`` re-registers a plugin of the same name, which is what makes 

72 editing a plugin in a notebook and re-running the cell work. 

73 """ 

74 if not plugin.name: 

75 raise PluginRegistrationError(f"{type(plugin).__name__} must set a `name`") 

76 if plugin.name in self._plugins: 

77 if not replace: 

78 raise PluginRegistrationError(f"A plugin named {plugin.name!r} is already loaded") 

79 self.unregister(plugin.name) 

80 

81 check_version(plugin.requires_cortex) 

82 

83 hooks = [ 

84 RegisteredHook(plugin.name, plugin.version, spec, callback) 

85 for spec, callback in plugin.declared_hooks() 

86 ] 

87 operations = [ 

88 RegisteredOperation(plugin.name, plugin.version, spec, callback) 

89 for spec, callback in plugin.declared_operations() 

90 ] 

91 if not hooks and not operations: 

92 raise PluginRegistrationError( 

93 f"Plugin {plugin.name!r} declares neither a hook nor an operation" 

94 ) 

95 

96 # Publish first, commit second. A plugin whose second operation clashes 

97 # with an existing one must leave nothing behind — otherwise the later 

98 # cleanup withdraws operations this plugin never owned, and a built-in 

99 # one disappears with it. 

100 published: list[RegisteredOperation] = [] 

101 try: 

102 for registered in operations: 

103 _publish_operation(registered) 

104 published.append(registered) 

105 except Exception: 

106 for registered in published: 

107 _withdraw_operation(registered) 

108 raise 

109 

110 self._plugins[plugin.name] = plugin 

111 self._hooks.extend(hooks) 

112 self._operations.extend(operations) 

113 

114 logger.info( 

115 "cortex.plugin_registered", 

116 plugin=plugin.name, 

117 version=plugin.version, 

118 hooks=len(hooks), 

119 operations=len(operations), 

120 ) 

121 

122 def unregister(self, name: str) -> None: 

123 """Remove a plugin and everything it declared.""" 

124 if name not in self._plugins: 

125 return 

126 for registered in [op for op in self._operations if op.plugin_name == name]: 

127 _withdraw_operation(registered) 

128 self._hooks = [h for h in self._hooks if h.plugin_name != name] 

129 self._operations = [o for o in self._operations if o.plugin_name != name] 

130 del self._plugins[name] 

131 logger.info("cortex.plugin_unregistered", plugin=name) 

132 

133 def clear(self) -> None: 

134 """Remove every plugin (shutdown, and test isolation).""" 

135 for name in list(self._plugins): 

136 self.unregister(name) 

137 

138 

139_registry: PluginRegistry | None = None 

140 

141 

142def get_plugin_registry() -> PluginRegistry: 

143 global _registry 

144 if _registry is None: 

145 _registry = PluginRegistry() 

146 return _registry 

147 

148 

149def _publish_operation(registered: RegisteredOperation) -> None: 

150 """Make a plugin operation reachable over HTTP and visible in /metadata.""" 

151 from ..api.operations import register_plugin_operation 

152 

153 register_plugin_operation(registered) 

154 

155 

156def _withdraw_operation(registered: RegisteredOperation) -> None: 

157 from ..api.operations import unregister_plugin_operation 

158 

159 unregister_plugin_operation(registered)