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

1"""Base protocols and data structures for export destinations. 

2 

3This module defines the core interfaces and data structures for the 

4NexusLIMS export framework, which allows records to be exported to 

5multiple repository destinations (CDCS, LabArchives, eLabFTW, etc.) 

6using a plugin-based architecture. 

7""" 

8 

9from __future__ import annotations 

10 

11import logging 

12from dataclasses import dataclass, field 

13from datetime import datetime 

14from typing import TYPE_CHECKING, Any, Protocol 

15 

16if TYPE_CHECKING: 

17 from pathlib import Path 

18 

19 from nexusLIMS.harvesters.reservation_event import ReservationEvent 

20 from nexusLIMS.schemas.activity import AcquisitionActivity 

21 

22_logger = logging.getLogger(__name__) 

23 

24 

25@dataclass 

26class ExportResult: 

27 """Result of a single export attempt. 

28 

29 Parameters 

30 ---------- 

31 success 

32 Whether the export succeeded 

33 destination_name 

34 Name of the destination plugin (e.g., "cdcs") 

35 record_id 

36 Destination-specific record identifier (if successful) 

37 record_url 

38 Direct URL to view the exported record (if successful) 

39 error_message 

40 Error message if export failed 

41 timestamp 

42 When the export attempt occurred 

43 metadata 

44 Additional destination-specific metadata 

45 """ 

46 

47 success: bool 

48 destination_name: str 

49 record_id: str | None = None 

50 record_url: str | None = None 

51 error_message: str | None = None 

52 timestamp: datetime = field(default_factory=datetime.now) 

53 metadata: dict[str, Any] = field(default_factory=dict) 

54 

55 def __repr__(self): 

56 """Return string representation of ExportResult.""" 

57 status = "SUCCESS" if self.success else "FAILED" 

58 return ( 

59 f"ExportResult(destination={self.destination_name}, " 

60 f"status={status}, " 

61 f"record_id={self.record_id})" 

62 ) 

63 

64 

65@dataclass 

66class ExportContext: 

67 """Context passed to export destination plugins. 

68 

69 Provides all necessary information for a destination to export a record, 

70 including file path, session metadata, and results from previously-run 

71 higher-priority destinations (for inter-destination dependencies). 

72 

73 Parameters 

74 ---------- 

75 xml_file_path 

76 Path to the XML record file to export 

77 session_identifier 

78 Unique identifier for this session 

79 instrument_pid 

80 Instrument identifier (e.g., "FEI-Titan-TEM-012345") 

81 dt_from 

82 Session start datetime 

83 dt_to 

84 Session end datetime 

85 user 

86 Username associated with this session (if known) 

87 metadata 

88 Additional session metadata 

89 previous_results 

90 Results from higher-priority destinations that have already run. 

91 Destinations can access these to create inter-destination 

92 dependencies (e.g., LabArchives including a CDCS link). 

93 """ 

94 

95 xml_file_path: Path 

96 session_identifier: str 

97 instrument_pid: str 

98 dt_from: datetime 

99 dt_to: datetime 

100 user: str | None = None 

101 metadata: dict[str, Any] = field(default_factory=dict) 

102 previous_results: dict[str, ExportResult] = field(default_factory=dict) 

103 activities: list[AcquisitionActivity] = field(default_factory=list) 

104 reservation_event: ReservationEvent | None = None 

105 

106 def get_result(self, destination_name: str) -> ExportResult | None: 

107 """Get result from a specific destination, if it has already run. 

108 

109 Parameters 

110 ---------- 

111 destination_name 

112 Name of the destination to query (e.g., "cdcs") 

113 

114 Returns 

115 ------- 

116 ExportResult | None 

117 The result if the destination has run, None otherwise 

118 """ 

119 return self.previous_results.get(destination_name) 

120 

121 def add_result(self, destination_name: str, result: ExportResult) -> None: 

122 """Add or update result from a destination. 

123 

124 Parameters 

125 ---------- 

126 destination_name 

127 Name of the destination (e.g., "cdcs", "elabftw") 

128 result 

129 The export result to store 

130 

131 Examples 

132 -------- 

133 >>> from nexusLIMS.exporters.base import ExportResult 

134 >>> result = ExportResult(success=True, message="Uploaded successfully") 

135 >>> context.add_result("cdcs", result) 

136 """ 

137 self.previous_results[destination_name] = result 

138 

139 def has_successful_export(self, destination_name: str) -> bool: 

140 """Check if a destination successfully exported. 

141 

142 Parameters 

143 ---------- 

144 destination_name 

145 Name of the destination to check (e.g., "cdcs") 

146 

147 Returns 

148 ------- 

149 bool 

150 True if the destination ran and succeeded, False otherwise 

151 """ 

152 result = self.get_result(destination_name) 

153 return result is not None and result.success 

154 

155 

156class ExportDestination(Protocol): 

157 """Protocol for export destination plugins. 

158 

159 Export destinations are discovered automatically by walking the 

160 exporters/destinations/ directory. Any class matching this protocol 

161 will be registered as an export destination. 

162 

163 Attributes 

164 ---------- 

165 name : str 

166 Unique identifier for this destination (e.g., "cdcs") 

167 priority : int 

168 Selection priority (0-1000, higher runs first). 

169 Use priority to manage inter-destination dependencies: 

170 higher-priority destinations run first and their results 

171 are available to lower-priority destinations. 

172 """ 

173 

174 name: str 

175 priority: int 

176 

177 @property 

178 def enabled(self) -> bool: 

179 """Whether this destination is enabled and configured. 

180 

181 Check if all required configuration is present (API keys, 

182 URLs, etc.) and the destination should be used for exports. 

183 

184 Returns 

185 ------- 

186 bool 

187 True if destination is ready to use, False otherwise 

188 """ 

189 ... 

190 

191 def validate_config(self) -> tuple[bool, str | None]: 

192 """Validate configuration. 

193 

194 Perform startup-time validation of configuration (API keys, 

195 connectivity, etc.) and return detailed error information. 

196 

197 Returns 

198 ------- 

199 tuple[bool, str | None] 

200 (is_valid, error_message) 

201 - is_valid: True if configuration is valid 

202 - error_message: None if valid, descriptive error if invalid 

203 """ 

204 ... 

205 

206 def export(self, context: ExportContext) -> ExportResult: 

207 """Export record to this destination. 

208 

209 CRITICAL: This method MUST NOT raise exceptions. All errors must 

210 be caught and returned as ExportResult with success=False and 

211 error_message set. 

212 

213 Parameters 

214 ---------- 

215 context 

216 Export context with file path, session metadata, and 

217 results from higher-priority destinations 

218 

219 Returns 

220 ------- 

221 ExportResult 

222 Result of the export attempt (success or failure) 

223 """ 

224 ...