Coverage for nexusLIMS/exporters/registry.py: 100%

1"""Plugin registry for export destinations. 

2 

3This module provides the ExporterRegistry singleton that auto-discovers 

4and manages export destination plugins from the destinations/ directory. 

5""" 

6 

7from __future__ import annotations 

8 

9import importlib 

10import inspect 

11import logging 

12import pkgutil 

13from pathlib import Path 

14from typing import TYPE_CHECKING, Literal 

15 

16from nexusLIMS.exporters.strategies import execute_strategy 

17 

18if TYPE_CHECKING: 

19 from nexusLIMS.exporters.base import ExportContext, ExportDestination, ExportResult 

20 

21_logger = logging.getLogger(__name__) 

22 

23#: Export strategy type: controls how records are exported to multiple destinations. 

24#: 

25#: - ``"all"``: Export to all destinations (default) 

26#: - ``"first_success"``: Stop after first successful export 

27#: - ``"best_effort"``: Continue exporting even if some destinations fail 

28ExportStrategy = Literal["all", "first_success", "best_effort"] 

29 

30 

31class ExporterRegistry: 

32 """Singleton registry for export destination plugins. 

33 

34 Auto-discovers plugins from exporters/destinations/ directory by 

35 examining all modules for classes that match the ExportDestination 

36 protocol (duck typing). 

37 

38 Attributes 

39 ---------- 

40 _destinations : dict[str, ExportDestination] 

41 Registered destination plugins, keyed by name 

42 _discovered : bool 

43 Whether plugin discovery has been performed 

44 """ 

45 

46 def __init__(self): 

47 """Initialize an empty registry.""" 

48 self._destinations: dict[str, ExportDestination] = {} 

49 self._discovered = False 

50 

51 def discover_plugins(self) -> None: 

52 """Auto-discover plugins from exporters/destinations/ directory. 

53 

54 Walks the destinations/ directory and examines all Python modules 

55 for classes matching the ExportDestination protocol. Discovered 

56 plugins are instantiated and registered by name. 

57 """ 

58 if self._discovered: 

59 return 

60 

61 _logger.info("Discovering export destination plugins...") 

62 

63 # Get path to destinations directory 

64 destinations_path = Path(__file__).parent / "destinations" 

65 if not destinations_path.exists(): 

66 _logger.warning( 

67 "Destinations directory not found: %s", 

68 destinations_path, 

69 ) 

70 self._discovered = True 

71 return 

72 

73 # Walk all modules in destinations directory 

74 for module_info in pkgutil.iter_modules([str(destinations_path)]): 

75 module_name = f"nexusLIMS.exporters.destinations.{module_info.name}" 

76 try: 

77 module = importlib.import_module(module_name) 

78 self._register_from_module(module) 

79 except Exception: 

80 _logger.exception( 

81 "Failed to load destination module: %s", 

82 module_name, 

83 ) 

84 continue 

85 

86 self._discovered = True 

87 _logger.info( 

88 "Discovered %d export destination(s): %s", 

89 len(self._destinations), 

90 ", ".join(self._destinations.keys()), 

91 ) 

92 

93 def _register_from_module(self, module) -> None: 

94 """Register plugins from a module. 

95 

96 Parameters 

97 ---------- 

98 module 

99 Python module to scan for ExportDestination implementations 

100 """ 

101 for name, obj in inspect.getmembers(module, inspect.isclass): 

102 # Skip imported classes from other modules 

103 if obj.__module__ != module.__name__: 

104 continue 

105 

106 # Check if class matches ExportDestination protocol 

107 if self._matches_protocol(obj): 

108 try: 

109 instance = obj() 

110 self._destinations[instance.name] = instance 

111 _logger.debug( 

112 "Registered export destination: %s (priority=%d)", 

113 instance.name, 

114 instance.priority, 

115 ) 

116 except Exception: 

117 _logger.exception( 

118 "Failed to instantiate destination plugin: %s", 

119 name, 

120 ) 

121 

122 def _matches_protocol(self, cls) -> bool: 

123 """Check if a class matches the ExportDestination protocol. 

124 

125 Uses duck typing to check for required attributes and methods: 

126 - name (attribute) 

127 - priority (attribute) 

128 - enabled (property) 

129 - validate_config (method) 

130 - export (method) 

131 

132 Parameters 

133 ---------- 

134 cls 

135 Class to check 

136 

137 Returns 

138 ------- 

139 bool 

140 True if class matches protocol, False otherwise 

141 """ 

142 # Check for required attributes (must be class attributes, not instance) 

143 try: 

144 # Check if name and priority exist as class-level attributes, 

145 # not just in __init__ 

146 if not hasattr(cls, "name") or not hasattr(cls, "priority"): 

147 return False 

148 

149 # Check for required methods 

150 required_methods = ["enabled", "validate_config", "export"] 

151 return all(hasattr(cls, method_name) for method_name in required_methods) 

152 except Exception: 

153 return False 

154 

155 def get_enabled_destinations(self) -> list[ExportDestination]: 

156 """Get enabled destinations sorted by priority (descending). 

157 

158 Returns only destinations where .enabled is True, sorted by 

159 priority (higher priority first). 

160 

161 Returns 

162 ------- 

163 list[ExportDestination] 

164 Enabled destinations in priority order 

165 """ 

166 self.discover_plugins() 

167 enabled = [d for d in self._destinations.values() if d.enabled] 

168 return sorted(enabled, key=lambda d: d.priority, reverse=True) 

169 

170 def export_to_all( 

171 self, 

172 context: ExportContext, 

173 *, 

174 strategy: ExportStrategy = "all", 

175 ) -> list[ExportResult]: 

176 """Export to destinations according to strategy. 

177 

178 Parameters 

179 ---------- 

180 context 

181 Export context with file path and session metadata 

182 strategy 

183 Export strategy to use (default: "all") 

184 

185 Returns 

186 ------- 

187 list[ExportResult] 

188 Results from each destination that was attempted 

189 """ 

190 return execute_strategy(strategy, self.get_enabled_destinations(), context) 

191 

192 

193# Singleton instance stored in a dict to avoid using `global` statement 

194_registry_holder: dict[str, ExporterRegistry] = {} 

195 

196 

197def get_registry() -> ExporterRegistry: 

198 """Get the global ExporterRegistry singleton. 

199 

200 Returns 

201 ------- 

202 ExporterRegistry 

203 The singleton registry instance 

204 """ 

205 if "instance" not in _registry_holder: 

206 _registry_holder["instance"] = ExporterRegistry() 

207 return _registry_holder["instance"]