Coverage for nexusLIMS/exporters/destinations/elabftw.py: 100%
103 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""eLabFTW export destination plugin.
3Exports NexusLIMS XML records to eLabFTW electronic lab notebook by creating
4experiments with markdown summaries and attaching the full XML record.
5"""
7from __future__ import annotations
9import logging
10import re
11from typing import Literal
13from pydantic import BaseModel, Field
15from nexusLIMS.config import settings
16from nexusLIMS.exporters.base import ExportContext, ExportResult
17from nexusLIMS.utils.elabftw import (
18 ContentType,
19 ELabFTWAuthenticationError,
20 get_elabftw_client,
21)
23_logger = logging.getLogger(__name__)
26# ============================================================================
27# eLabFTW Extra Fields Schema Models
28# ============================================================================
31class ExtraFieldsGroup(BaseModel):
32 """eLabFTW extra fields group definition.
34 Groups are used to organize related fields in the eLabFTW UI.
35 """
37 id: int = Field(..., description="Unique group ID")
38 name: str = Field(..., description="Display name of the group")
41class ExtraField(BaseModel):
42 """eLabFTW extra field definition.
44 Represents a single structured metadata field with type validation.
45 See: https://doc.elabftw.net/metadata.html#schema-description
46 """
48 type: Literal[
49 "text",
50 "date",
51 "datetime-local",
52 "email",
53 "number",
54 "select",
55 "radio",
56 "checkbox",
57 "url",
58 "time",
59 ] = Field(..., description="Field type for validation and UI rendering")
60 value: str | int | float | bool = Field(..., description="Field value")
61 description: str | None = Field(None, description="Help text for the field")
62 position: int | None = Field(None, description="Display order (lower first)")
63 group_id: int | None = Field(
64 None, description="ID of the group this field belongs to"
65 )
66 required: bool | None = Field(None, description="Whether field is required")
67 blank_value_on_duplicate: bool | None = Field(
68 None, description="Clear value when entity is duplicated"
69 )
71 model_config = {"extra": "allow"} # Allow additional eLabFTW-specific fields
74class ELabFTWConfig(BaseModel):
75 """eLabFTW configuration object for extra fields metadata."""
77 display_main_text: bool = Field(
78 default=True, description="Whether to display the main text/body"
79 )
80 extra_fields_groups: list[ExtraFieldsGroup] = Field(
81 default_factory=list, description="Group definitions for organizing fields"
82 )
85class ExtraFieldsMetadata(BaseModel):
86 """Complete eLabFTW extra fields metadata structure.
88 This is the top-level object sent to eLabFTW's metadata field.
89 """
91 extra_fields: dict[str, ExtraField] = Field(
92 ..., description="Field definitions keyed by field name"
93 )
94 elabftw: ELabFTWConfig = Field(..., description="eLabFTW-specific configuration")
97class ELabFTWDestination:
98 """eLabFTW export destination plugin.
100 Creates one eLabFTW experiment per NexusLIMS session, with a markdown
101 summary of the session and the full XML record attached as a file.
103 Attributes
104 ----------
105 name : str
106 Destination identifier: "elabftw"
107 priority : int
108 Export priority: 85 (after CDCS but before LabArchives)
109 """
111 name = "elabftw"
112 priority = 85 # After CDCS (100), before LabArchives (90)
114 @property
115 def enabled(self) -> bool:
116 """Check if eLabFTW is configured and enabled.
118 Returns
119 -------
120 bool
121 True if both NX_ELABFTW_API_KEY and NX_ELABFTW_URL are configured
122 """
123 return (
124 settings.NX_ELABFTW_API_KEY is not None
125 and settings.NX_ELABFTW_API_KEY != ""
126 and settings.NX_ELABFTW_URL is not None
127 and settings.NX_ELABFTW_URL != ""
128 )
130 def validate_config(self) -> tuple[bool, str | None]:
131 """Validate eLabFTW configuration.
133 Tests:
134 - NX_ELABFTW_API_KEY is configured
135 - NX_ELABFTW_URL is configured
136 - Can authenticate to eLabFTW API
138 Returns
139 -------
140 tuple[bool, str | None]
141 (is_valid, error_message)
142 """
143 if not settings.NX_ELABFTW_API_KEY:
144 return False, "NX_ELABFTW_API_KEY not configured"
146 if not settings.NX_ELABFTW_URL:
147 return False, "NX_ELABFTW_URL not configured"
149 # Test authentication by listing experiments (limit 1)
150 try:
151 client = get_elabftw_client()
152 client.list_experiments(limit=1)
153 except ELabFTWAuthenticationError as e:
154 return False, f"eLabFTW authentication failed: {e}"
155 except Exception as e:
156 return False, f"eLabFTW configuration error: {e}"
158 return True, None
160 def export(self, context: ExportContext) -> ExportResult:
161 """Export record to eLabFTW.
163 Creates an experiment with an HTML summary of the session,
164 then attaches the XML record file. Never raises exceptions - all
165 errors are caught and returned as ExportResult with success=False.
167 Parameters
168 ----------
169 context
170 Export context with file path and session metadata
172 Returns
173 -------
174 ExportResult
175 Result of the export attempt
176 """
177 try:
178 # Get eLabFTW client
179 client = get_elabftw_client()
181 # Build experiment content
182 title = self._build_title(context)
183 body = self._build_html_body(context)
184 tags = self._build_tags(context)
185 metadata = self._build_metadata(context)
187 # Create experiment
188 # Note: Using HTML instead of Markdown due to eLabFTW API bug
189 # https://github.com/elabftw/elabftw/issues/6416
190 experiment = client.create_experiment(
191 title=title,
192 body=body,
193 tags=tags,
194 metadata=metadata,
195 category=settings.NX_ELABFTW_EXPERIMENT_CATEGORY,
196 status=settings.NX_ELABFTW_EXPERIMENT_STATUS,
197 content_type=ContentType.HTML,
198 )
200 experiment_id = experiment["id"]
201 _logger.info("Created eLabFTW experiment %s: %s", experiment_id, title)
203 # Upload XML file as attachment
204 client.upload_file_to_experiment(
205 experiment_id=experiment_id,
206 file_path=context.xml_file_path,
207 comment="NexusLIMS XML record",
208 )
210 # Build experiment URL
211 experiment_url = (
212 f"{settings.NX_ELABFTW_URL}/"
213 f"experiments.php?mode=view&id={experiment_id}"
214 )
216 return ExportResult(
217 success=True,
218 destination_name=self.name,
219 record_id=str(experiment_id),
220 record_url=experiment_url,
221 )
223 except Exception as e:
224 _logger.exception(
225 "Failed to export to eLabFTW: %s",
226 context.xml_file_path.name,
227 )
228 return ExportResult(
229 success=False,
230 destination_name=self.name,
231 error_message=str(e),
232 )
234 def _build_title(self, context: ExportContext) -> str:
235 """Build experiment title.
237 Parameters
238 ----------
239 context
240 Export context
242 Returns
243 -------
244 str
245 Title in format: "NexusLIMS - {instrument} - {session_id}"
246 """
247 return f"NexusLIMS Experiment - {context.xml_file_path.stem}"
249 def _build_html_body(self, context: ExportContext) -> str:
250 """Build HTML body for experiment.
252 Note: We use HTML instead of Markdown due to an eLabFTW API bug that
253 prevents setting content_type via the API.
254 See: https://github.com/elabftw/elabftw/issues/6416
256 Parameters
257 ----------
258 context
259 Export context
261 Returns
262 -------
263 str
264 HTML-formatted body with session details and CDCS link
265 """
266 lines = [
267 "<h1>NexusLIMS Microscopy Session</h1>",
268 "<h2>Session Details</h2>",
269 "<ul>",
270 f"<li><strong>Session ID</strong>: {context.session_identifier}</li>",
271 f"<li><strong>Instrument</strong>: {context.instrument_pid}</li>",
272 ]
274 # Add user if available
275 if context.user:
276 lines.append(f"<li><strong>User</strong>: {context.user}</li>")
278 # Add timestamps
279 lines.extend(
280 [
281 f"<li><strong>Start</strong>: {context.dt_from.isoformat()}</li>",
282 f"<li><strong>End</strong>: {context.dt_to.isoformat()}</li>",
283 "</ul>",
284 ]
285 )
287 # Add CDCS link if available
288 cdcs_result = context.get_result("cdcs")
289 if cdcs_result and cdcs_result.success and cdcs_result.record_url:
290 lines.extend(
291 [
292 "<h2>Related Records</h2>",
293 "<ul>",
294 f'<li><a href="{cdcs_result.record_url}">View in CDCS</a></li>',
295 "</ul>",
296 ]
297 )
299 # Add note about XML attachment
300 lines.extend(
301 [
302 "<h2>Files</h2>",
303 (
304 "<p>The complete NexusLIMS XML record is attached to "
305 "this experiment.</p>"
306 ),
307 ]
308 )
310 return "\n".join(lines)
312 def _build_tags(self, context: ExportContext) -> list[str]:
313 """Build tag list for experiment.
315 Parameters
316 ----------
317 context
318 Export context
320 Returns
321 -------
322 list of str
323 Tags including "NexusLIMS", instrument, and user
324 """
325 tags = ["NexusLIMS", context.instrument_pid]
327 if context.user:
328 tags.append(context.user)
330 return tags
332 def _build_metadata(self, context: ExportContext) -> dict:
333 """Build metadata using eLabFTW extra_fields schema.
335 This method creates a structured metadata object following eLabFTW's
336 extra_fields format, which provides type validation, field descriptions,
337 grouping, and ordering.
339 Parameters
340 ----------
341 context
342 Export context with session information
344 Returns
345 -------
346 dict
347 Metadata dict with 'extra_fields' and 'elabftw' keys conforming to
348 eLabFTW's extra_fields schema. See:
349 https://doc.elabftw.net/metadata.html#schema-description
351 Notes
352 -----
353 The extra_fields schema provides several benefits over flat metadata:
354 - Type validation (datetime-local, url, text, etc.)
355 - Field descriptions for documentation
356 - Logical grouping of related fields
357 - Controlled ordering via position attribute
358 - Better UI/UX in eLabFTW interface
359 """
360 # Build extra fields using Pydantic models for type safety
361 extra_fields: dict[str, ExtraField] = {
362 "Session ID": ExtraField(
363 type="text",
364 value=context.session_identifier,
365 description="NexusLIMS session identifier",
366 position=1,
367 group_id=1,
368 ),
369 "Instrument": ExtraField(
370 type="text",
371 value=context.instrument_pid,
372 description="Instrument persistent identifier",
373 position=2,
374 group_id=1,
375 ),
376 "Start Time": ExtraField(
377 type="datetime-local",
378 value=context.dt_from.strftime("%Y-%m-%dT%H:%M"),
379 description="Session start time",
380 position=3,
381 group_id=1,
382 ),
383 "End Time": ExtraField(
384 type="datetime-local",
385 value=context.dt_to.strftime("%Y-%m-%dT%H:%M"),
386 description="Session end time",
387 position=4,
388 group_id=1,
389 ),
390 }
392 # Add optional user field
393 if context.user:
394 extra_fields["User"] = ExtraField(
395 type="text",
396 value=context.user,
397 description="User who performed the session",
398 position=5,
399 group_id=1,
400 )
402 # Define groups
403 groups = [ExtraFieldsGroup(id=1, name="Session Information")]
405 # Add CDCS cross-link if available
406 cdcs_result = context.get_result("cdcs")
407 if cdcs_result and cdcs_result.success and cdcs_result.record_url:
408 extra_fields["CDCS Record"] = ExtraField(
409 type="url",
410 value=cdcs_result.record_url,
411 description="Link to CDCS record",
412 position=10, # Leave gap for potential future fields
413 group_id=2,
414 )
415 groups.append(ExtraFieldsGroup(id=2, name="Related Records"))
417 # Create and validate the complete metadata structure
418 metadata = ExtraFieldsMetadata(
419 extra_fields=extra_fields,
420 elabftw=ELabFTWConfig(
421 display_main_text=True,
422 extra_fields_groups=groups,
423 ),
424 )
426 # Return as dict for API compatibility
427 return metadata.model_dump(exclude_none=True)
429 def _validate_extra_field(self, field_name: str, field_def: dict) -> bool:
430 """Validate an extra_field definition.
432 Checks that the field has required keys (type, value) and that
433 the value matches the declared type format.
435 Parameters
436 ----------
437 field_name
438 Name of the field
439 field_def
440 Field definition dict with type, value, and optional metadata
442 Returns
443 -------
444 bool
445 True if valid, False otherwise
447 Notes
448 -----
449 Validation rules by type:
450 - datetime-local: YYYY-MM-DDTHH:MM format
451 - date: YYYY-MM-DD format
452 - url: must start with http:// or https://
453 - Other types: no format validation
454 """
455 required_keys = {"type", "value"}
456 if not all(key in field_def for key in required_keys):
457 _logger.warning(
458 "Extra field '%s' missing required keys: %s",
459 field_name,
460 required_keys,
461 )
462 return False
464 field_type = field_def["type"]
465 value = field_def["value"]
467 # Type-specific validation
468 if field_type == "datetime-local":
469 # Value should be in format YYYY-MM-DDTHH:MM
470 if not re.match(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", str(value)):
471 _logger.warning(
472 "Extra field '%s' has invalid datetime-local format: %s",
473 field_name,
474 value,
475 )
476 return False
477 elif field_type == "date":
478 # Value should be in format YYYY-MM-DD
479 if not re.match(r"\d{4}-\d{2}-\d{2}", str(value)):
480 _logger.warning(
481 "Extra field '%s' has invalid date format: %s",
482 field_name,
483 value,
484 )
485 return False
486 elif field_type == "url" and not str(value).startswith(("http://", "https://")):
487 # Basic URL validation
488 _logger.warning(
489 "Extra field '%s' has invalid URL: %s",
490 field_name,
491 value,
492 )
493 return False
495 return True