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

1"""Export strategies for multi-destination export. 

2 

3This module implements different strategies for exporting to multiple 

4destinations with different success criteria: 

5- all: All destinations must succeed 

6- first_success: Stop after first success 

7- best_effort: Try all, succeed if any succeed 

8""" 

9 

10from __future__ import annotations 

11 

12import logging 

13from typing import TYPE_CHECKING 

14 

15if TYPE_CHECKING: 

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

17 from nexusLIMS.exporters.registry import ExportStrategy 

18 

19_logger = logging.getLogger(__name__) 

20 

21 

22def execute_strategy( 

23 strategy: ExportStrategy, 

24 destinations: list[ExportDestination], 

25 context: ExportContext, 

26) -> list[ExportResult]: 

27 """Execute export strategy. 

28 

29 Exports to multiple destinations according to the specified strategy, 

30 accumulating results in context.previous_results to enable 

31 inter-destination dependencies. 

32 

33 Parameters 

34 ---------- 

35 strategy 

36 Export strategy to use 

37 destinations 

38 List of destinations (should be sorted by priority) 

39 context 

40 Export context (will be mutated to add previous_results) 

41 

42 Returns 

43 ------- 

44 list[ExportResult] 

45 Results from each destination that was attempted 

46 """ 

47 if strategy == "all": 

48 return _strategy_all(destinations, context) 

49 if strategy == "first_success": 

50 return _strategy_first_success(destinations, context) 

51 if strategy == "best_effort": 

52 return _strategy_best_effort(destinations, context) 

53 msg = f"Unknown export strategy: {strategy}" 

54 raise ValueError(msg) 

55 

56 

57def _strategy_all( 

58 destinations: list[ExportDestination], 

59 context: ExportContext, 

60) -> list[ExportResult]: 

61 """All destinations must succeed. 

62 

63 Export to all destinations. If any export fails, log a warning but 

64 continue to remaining destinations. The strategy is considered 

65 successful only if ALL destinations succeed. 

66 

67 Parameters 

68 ---------- 

69 destinations 

70 List of destinations to export to 

71 context 

72 Export context 

73 

74 Returns 

75 ------- 

76 list[ExportResult] 

77 Results from all destinations 

78 """ 

79 results = [] 

80 

81 for dest in destinations: 

82 _logger.info("Exporting to %s (priority=%d)...", dest.name, dest.priority) 

83 result = dest.export(context) 

84 results.append(result) 

85 

86 # Add result to context for subsequent destinations 

87 context.add_result(dest.name, result) 

88 

89 if not result.success: 

90 _logger.warning( 

91 "Export to %s failed (all strategy): %s", 

92 dest.name, 

93 result.error_message, 

94 ) 

95 else: 

96 _logger.info("Export to %s succeeded", dest.name) 

97 

98 # Check overall success 

99 success_count = sum(1 for r in results if r.success) 

100 if success_count == len(results): 

101 _logger.info("All %d destination(s) succeeded (all strategy)", len(results)) 

102 else: 

103 _logger.warning( 

104 "Only %d/%d destination(s) succeeded (all strategy)", 

105 success_count, 

106 len(results), 

107 ) 

108 

109 return results 

110 

111 

112def _strategy_first_success( 

113 destinations: list[ExportDestination], 

114 context: ExportContext, 

115) -> list[ExportResult]: 

116 """Stop after first success. 

117 

118 Export to destinations in priority order until one succeeds. 

119 Stop immediately after the first successful export. 

120 

121 Parameters 

122 ---------- 

123 destinations 

124 List of destinations to export to (should be priority-sorted) 

125 context 

126 Export context 

127 

128 Returns 

129 ------- 

130 list[ExportResult] 

131 Results from destinations that were attempted (up to first success) 

132 """ 

133 results = [] 

134 

135 for dest in destinations: 

136 _logger.info("Exporting to %s (priority=%d)...", dest.name, dest.priority) 

137 result = dest.export(context) 

138 results.append(result) 

139 

140 # Add result to context for subsequent destinations 

141 context.add_result(dest.name, result) 

142 

143 if result.success: 

144 _logger.info( 

145 "Export succeeded to %s, stopping (first_success strategy)", 

146 dest.name, 

147 ) 

148 break 

149 _logger.warning("Export to %s failed: %s", dest.name, result.error_message) 

150 

151 if not any(r.success for r in results): 

152 _logger.error( 

153 "All %d destination(s) failed (first_success strategy)", 

154 len(results), 

155 ) 

156 

157 return results 

158 

159 

160def _strategy_best_effort( 

161 destinations: list[ExportDestination], 

162 context: ExportContext, 

163) -> list[ExportResult]: 

164 """Try all, succeed if any succeed. 

165 

166 Export to all destinations regardless of individual failures. 

167 The strategy is considered successful if at least one destination 

168 succeeds. 

169 

170 Parameters 

171 ---------- 

172 destinations 

173 List of destinations to export to 

174 context 

175 Export context 

176 

177 Returns 

178 ------- 

179 list[ExportResult] 

180 Results from all destinations 

181 """ 

182 results = [] 

183 

184 for dest in destinations: 

185 _logger.info("Exporting to %s (priority=%d)...", dest.name, dest.priority) 

186 result = dest.export(context) 

187 results.append(result) 

188 

189 # Add result to context for subsequent destinations 

190 context.add_result(dest.name, result) 

191 

192 if result.success: 

193 _logger.info("Export to %s succeeded", dest.name) 

194 else: 

195 _logger.warning("Export to %s failed: %s", dest.name, result.error_message) 

196 

197 # Check overall success 

198 success_count = sum(1 for r in results if r.success) 

199 if success_count > 0: 

200 _logger.info( 

201 "Export succeeded to %d/%d destination(s) (best_effort strategy)", 

202 success_count, 

203 len(results), 

204 ) 

205 else: 

206 _logger.error( 

207 "All %d destination(s) failed (best_effort strategy)", 

208 len(results), 

209 ) 

210 

211 return results