Coverage for src/lumen/cortex/plugin.py: 97%

1""" 

2What a plugin author writes: the ``Plugin`` base class and the two decorators. 

3 

4There are exactly two extension points — a write-time hook and a custom 

5operation. That is deliberate: a model running inside the server process can 

6take the server down with it, so extensions attach at defined places with 

7timeouts rather than anywhere with none. 

8""" 

9 

10from __future__ import annotations 

11 

12import inspect 

13from collections.abc import Awaitable, Callable 

14from dataclasses import dataclass 

15from typing import Any, ClassVar, Literal, TypeVar 

16 

17from .contract import EVENT_RESOURCE_WRITTEN, KNOWN_EVENTS, PluginRegistrationError 

18 

19#: Default budget for a write-time hook. It runs inside the triggering request, 

20#: so the default is small on purpose; long work belongs in an operation. 

21DEFAULT_HOOK_TIMEOUT_MS = 1_000 

22#: Custom operations are invoked deliberately by a client and may take longer. 

23DEFAULT_OPERATION_TIMEOUT_MS = 30_000 

24 

25OperationScope = Literal["system", "type", "instance"] 

26 

27_HOOK_ATTR = "__cortex_hooks__" 

28_OPERATION_ATTR = "__cortex_operations__" 

29 

30F = TypeVar("F", bound=Callable[..., Awaitable[Any]]) 

31 

32 

33@dataclass(frozen=True) 

34class HookSpec: 

35 """A declared write-time hook, before it is bound to an instance.""" 

36 

37 event: str 

38 resource_type: str | None 

39 timeout_ms: int 

40 

41 def matches(self, event: str, resource_type: str) -> bool: 

42 return self.event == event and self.resource_type in (None, resource_type) 

43 

44 

45@dataclass(frozen=True) 

46class OperationSpec: 

47 """A declared custom operation, before it is bound to an instance.""" 

48 

49 #: Operation name without the leading ``$``. 

50 name: str 

51 scope: OperationScope 

52 resource_type: str | None 

53 timeout_ms: int 

54 #: Canonical ``OperationDefinition`` URL for the CapabilityStatement. 

55 definition: str | None 

56 

57 

58def hook( 

59 event: str = EVENT_RESOURCE_WRITTEN, 

60 *, 

61 resource_type: str | None = None, 

62 timeout_ms: int = DEFAULT_HOOK_TIMEOUT_MS, 

63) -> Callable[[F], F]: 

64 """Run this method after a write to ``resource_type`` has committed. 

65 

66 The method is called as ``method(ctx, resource)`` and its return value is 

67 ignored — a hook reacts, it does not alter the resource that triggered it. 

68 Exceeding ``timeout_ms`` cancels the hook; the triggering write stands. 

69 """ 

70 if event not in KNOWN_EVENTS: 

71 raise PluginRegistrationError( 

72 f"Unknown hook event {event!r}; known events: {sorted(KNOWN_EVENTS)}" 

73 ) 

74 if timeout_ms <= 0: 

75 raise PluginRegistrationError("timeout_ms must be positive") 

76 

77 def decorator(fn: F) -> F: 

78 specs: list[HookSpec] = list(getattr(fn, _HOOK_ATTR, [])) 

79 specs.append(HookSpec(event=event, resource_type=resource_type, timeout_ms=timeout_ms)) 

80 setattr(fn, _HOOK_ATTR, specs) 

81 return fn 

82 

83 return decorator 

84 

85 

86def operation( 

87 name: str, 

88 *, 

89 on: str | None = None, 

90 scope: OperationScope | None = None, 

91 timeout_ms: int = DEFAULT_OPERATION_TIMEOUT_MS, 

92 definition: str | None = None, 

93) -> Callable[[F], F]: 

94 """Expose this method as a FHIR custom operation. 

95 

96 ``on`` names the resource type the operation hangs off; without it the 

97 operation is registered at system level. The scope defaults to ``instance`` 

98 when ``on`` is given (``POST /Patient/123/$name``) and can be set to 

99 ``type`` explicitly (``POST /Patient/$name``). 

100 

101 The method is called as ``method(ctx, op)`` where ``op`` carries the 

102 invocation (resource id, request body, query parameters), and returns 

103 either a FHIR resource as a ``dict`` or ``None`` for an empty 200. 

104 """ 

105 resolved_scope: OperationScope = scope or ("instance" if on else "system") 

106 if resolved_scope in ("type", "instance") and not on: 

107 raise PluginRegistrationError(f"Operation ${name} at {resolved_scope} level needs `on=`") 

108 if timeout_ms <= 0: 

109 raise PluginRegistrationError("timeout_ms must be positive") 

110 

111 def decorator(fn: F) -> F: 

112 specs: list[OperationSpec] = list(getattr(fn, _OPERATION_ATTR, [])) 

113 specs.append( 

114 OperationSpec( 

115 name=name.lstrip("$"), 

116 scope=resolved_scope, 

117 resource_type=on, 

118 timeout_ms=timeout_ms, 

119 definition=definition, 

120 ) 

121 ) 

122 setattr(fn, _OPERATION_ATTR, specs) 

123 return fn 

124 

125 return decorator 

126 

127 

128class Plugin: 

129 """Base class for everything that plugs into Lumen. 

130 

131 Subclasses set ``name`` and may set ``version`` and ``requires_cortex``, 

132 then declare methods carrying :func:`hook` or :func:`operation`. 

133 """ 

134 

135 #: Stable identifier. Appears in logs, metrics and in the Provenance of 

136 #: anything the plugin writes, so it should not change between versions. 

137 name: ClassVar[str] = "" 

138 version: ClassVar[str] = "0.0.0" 

139 #: Minimum Cortex surface this plugin needs, e.g. ``"1.0"``. 

140 requires_cortex: ClassVar[str | None] = None 

141 

142 async def setup(self) -> None: 

143 """Called once when the plugin is registered. Load models here.""" 

144 

145 async def teardown(self) -> None: 

146 """Called once when the plugin is removed or the server shuts down.""" 

147 

148 # ── Declaration discovery ───────────────────────────────────────────────── 

149 

150 def declared_hooks(self) -> list[tuple[HookSpec, Callable[..., Awaitable[Any]]]]: 

151 return _declarations(self, _HOOK_ATTR) 

152 

153 def declared_operations(self) -> list[tuple[OperationSpec, Callable[..., Awaitable[Any]]]]: 

154 return _declarations(self, _OPERATION_ATTR) 

155 

156 

157def _declarations(plugin: Plugin, attr: str) -> list[tuple[Any, Callable[..., Awaitable[Any]]]]: 

158 """Bound methods of ``plugin`` carrying declarations under ``attr``.""" 

159 found: list[tuple[Any, Callable[..., Awaitable[Any]]]] = [] 

160 for _, member in inspect.getmembers(type(plugin), callable): 

161 specs: list[Any] = list(getattr(member, attr, [])) 

162 if not specs: 

163 continue 

164 bound = getattr(plugin, member.__name__) 

165 if not inspect.iscoroutinefunction(member): 

166 raise PluginRegistrationError( 

167 f"{type(plugin).__name__}.{member.__name__} must be an async def" 

168 ) 

169 for spec in specs: 

170 found.append((spec, bound)) 

171 return found