Coverage for src/lumen/cortex/runtime.py: 96%

1""" 

2Registering a plugin while the server is already running. 

3 

4This is a **development** path. It exists because the shortest way to show what 

5Cortex is — and the fastest way to write a plugin — is to edit a cell in a 

6notebook attached to the running server and re-run it, instead of restarting 

7with a changed environment variable. 

8 

9It is off unless ``LUMEN_CORTEX_ALLOW_RUNTIME_REGISTRATION`` is set, because 

10anything that can call this function can put arbitrary code on the write path 

11of a live FHIR server. In production, plugins come from configuration and are 

12loaded once at startup. 

13 

14There is deliberately no HTTP endpoint for this. Registration happens from 

15inside the process or not at all. 

16""" 

17 

18from __future__ import annotations 

19 

20import structlog 

21 

22from ..config import get_settings 

23from .contract import PluginRegistrationError 

24from .plugin import Plugin 

25from .registry import get_plugin_registry 

26 

27logger = structlog.get_logger(__name__) 

28 

29 

30async def register_plugin(plugin: Plugin | type[Plugin], *, replace: bool = True) -> Plugin: 

31 """Attach a plugin to the running server. Returns the registered instance. 

32 

33 Re-running the same registration replaces the previous version of the 

34 plugin, so editing and re-executing a notebook cell does what you expect. 

35 """ 

36 settings = get_settings() 

37 if not settings.cortex.enabled: 

38 raise PluginRegistrationError( 

39 "Cortex is disabled on this server. Set LUMEN_CORTEX_ENABLED=true." 

40 ) 

41 if not settings.cortex.allow_runtime_registration: 

42 raise PluginRegistrationError( 

43 "Runtime plugin registration is disabled. Set " 

44 "LUMEN_CORTEX_ALLOW_RUNTIME_REGISTRATION=true for development, or " 

45 "list the plugin in LUMEN_CORTEX_PLUGINS for production." 

46 ) 

47 

48 instance = plugin() if isinstance(plugin, type) else plugin 

49 await instance.setup() 

50 get_plugin_registry().register(instance, replace=replace) 

51 await logger.ainfo("cortex.plugin_registered_at_runtime", plugin=instance.name) 

52 return instance 

53 

54 

55async def unregister_plugin(name: str) -> None: 

56 """Detach a plugin from the running server.""" 

57 registry = get_plugin_registry() 

58 for plugin in registry.plugins: 

59 if plugin.name == name: 

60 await plugin.teardown() 

61 registry.unregister(name)