Coverage for nexusLIMS/utils/elabftw.py: 100%

1"""Low-level API client for eLabFTW electronic lab notebook.

3This module provides a reusable client for interacting with eLabFTW's REST API v2.

4It handles authentication, request/response formatting, and error handling for

5CRUD operations on experiments.

7eLabFTW API Documentation:

8 https://doc.elabftw.net/api/v2/

10Example usage:

11 >>> from nexusLIMS.utils.elabftw import get_elabftw_client

12 >>> client = get_elabftw_client()

13 >>> exp = client.create_experiment(

14 ... title="My Experiment",

15 ... body="Experiment description",

16 ... tags=["microscopy", "nexuslims"]

17 ... )

18 >>> print(f"Created experiment {exp['id']}")

19"""

21import json

22import logging

23from enum import IntEnum

24from http import HTTPStatus

25from pathlib import Path

26from typing import Any

28from nexusLIMS.config import settings

29from nexusLIMS.utils.network import nexus_req

31_logger = logging.getLogger(__name__)

34class ELabFTWError(Exception):

35 """Base exception for eLabFTW API errors."""

38class ELabFTWAuthenticationError(ELabFTWError):

39 """Authentication failed (invalid or missing API key)."""

42class ELabFTWNotFoundError(ELabFTWError):

43 """Requested resource not found (404)."""

46class ContentType(IntEnum):

47 """eLabFTW content type for experiment body text.

49 Specifies how the body text of an experiment should be interpreted

50 and rendered by eLabFTW.

52 Attributes

53 ----------

54 HTML : int

55 Body text is HTML formatted (value: 1, default in eLabFTW)

56 MARKDOWN : int

57 Body text is Markdown formatted (value: 2)

58 """

60 HTML = 1

61 MARKDOWN = 2

64class State(IntEnum):

65 """eLabFTW experiment state enumeration.

67 These states represent the lifecycle status of experiments in eLabFTW.

68 Values correspond to the eLabFTW database schema.

70 Attributes

71 ----------

72 Normal : int

73 Standard active experiment (value: 1)

74 Archived : int

75 Experiment has been archived (value: 2)

76 Deleted : int

77 Experiment has been soft-deleted (value: 3)

78 Pending : int

79 Experiment is pending approval or processing (value: 4)

80 Processing : int

81 Experiment is currently being processed (value: 5)

82 Error : int

83 Experiment encountered an error state (value: 6)

84 """

86 Normal = 1

87 Archived = 2

88 Deleted = 3

89 Pending = 4

90 Processing = 5

91 Error = 6

94class ELabFTWClient:

95 """Low-level client for eLabFTW API v2.

97 This client provides basic CRUD operations for eLabFTW experiments using

98 the REST API v2. It handles authentication via API key and provides

99 consistent error handling.

100

101 Parameters

102 ----------

103 base_url : str

104 Root URL of the eLabFTW instance (e.g., "https://elabftw.example.com").

105 Do not include the API path - it will be appended automatically.

106 api_key : str

107 API key from eLabFTW user panel. Must have write permissions for

108 creating/updating experiments.

109

110 Attributes

111 ----------

112 base_url : str

113 Root URL of eLabFTW instance

114 api_key : str

115 API authentication key

116 experiments_endpoint : str

117 Full URL to experiments API endpoint

118

119 Examples

120 --------

121 >>> client = ELabFTWClient(

122 ... base_url="https://elabftw.example.com",

123 ... api_key="your-api-key-here"

124 ... )

125 >>> experiments = client.list_experiments(limit=5)

126 >>> for exp in experiments:

127 ... print(f"{exp['id']}: {exp['title']}")

128 """

129

130 def __init__(self, base_url: str, api_key: str):

131 """Initialize eLabFTW API client.

132

133 Parameters

134 ----------

135 base_url : str

136 Root URL of eLabFTW instance

137 api_key : str

138 API key for authentication

139 """

140 self.base_url = base_url.rstrip("/")

141 self.api_key = api_key

142 self.experiments_endpoint = f"{self.base_url}/api/v2/experiments"

143

144 def _make_request(

145 self,

146 method: str,

147 url: str,

148 json_data: dict[str, Any] | None = None,

149 files: dict[str, Any] | None = None,

150 ) -> dict[str, Any] | None:

151 """Make authenticated HTTP request to eLabFTW API.

152

153 Parameters

154 ----------

155 method : str

156 HTTP method (GET, POST, PATCH, DELETE)

157 url : str

158 Full URL to request

159 json_data : dict, optional

160 JSON body for request

161 files : dict, optional

162 Files to upload (multipart/form-data)

163

164 Returns

165 -------

166 dict or None

167 Response JSON data, or None for 204 No Content

168

169 Raises

170 ------

171 ELabFTWAuthenticationError

172 If authentication fails (401)

173 ELabFTWNotFoundError

174 If resource not found (404)

175 ELabFTWError

176 For other API errors

177 """

178 headers = {"Authorization": self.api_key}

179

180 # Make the request - wrap any network/request exceptions

181 try:

182 response = nexus_req(

183 url,

184 method,

185 headers=headers,

186 json=json_data,

187 files=files,

188 )

189 except Exception as e:

190 msg = f"Request to eLabFTW API failed: {e}"

191 raise ELabFTWError(msg) from e

192

193 # Handle different success status codes

194 if response.status_code == HTTPStatus.NO_CONTENT:

195 # No Content - successful delete

196 return None

197

198 if response.status_code == HTTPStatus.CREATED:

199 # Created - eLabFTW returns Location header with experiment URL

200 # Response body is typically empty

201 location = response.headers.get("Location")

202 if location:

203 # Extract experiment ID by removing the endpoint URL

204 # E.g., "http://host/api/v2/experiments/123" -> "123"

205 try:

206 experiment_id = int(location.rstrip("/").rsplit("/", 1)[-1])

207 except ValueError as e:

208 msg = (

209 "Failed to parse experiment ID from "

210 f"Location header: {location}"

211 )

212 raise ELabFTWError(msg) from e

213 else:

214 return {"id": experiment_id, "location": location}

215 # Fallback: try to parse JSON response (in case API behavior changes)

216 try:

217 return response.json()

218 except Exception:

219 msg = "201 Created response missing Location header and JSON body"

220 raise ELabFTWError(msg) from None

221

222 if response.status_code == HTTPStatus.OK:

223 # Success with JSON response - wrap JSON parsing exceptions

224 try:

225 return response.json()

226 except Exception as e:

227 msg = f"Failed to parse response JSON: {e}"

228 raise ELabFTWError(msg) from e

229

230 # Handle error responses

231 if response.status_code == HTTPStatus.UNAUTHORIZED:

232 msg = "Authentication failed - check API key"

233 raise ELabFTWAuthenticationError(msg)

234

235 if response.status_code == HTTPStatus.NOT_FOUND:

236 msg = f"Resource not found: {url}"

237 raise ELabFTWNotFoundError(msg)

238

239 # Generic error for other status codes

240 msg = f"API request failed with status {response.status_code}: {response.text}"

241 raise ELabFTWError(msg)

242

243 def create_experiment( # noqa: PLR0913

244 self,

245 title: str,

246 body: str | None = None,

247 tags: list[str] | None = None,

248 metadata: dict[str, Any] | None = None,

249 category: int | None = None,

250 status: int | None = None,

251 content_type: ContentType | None = None,

252 ) -> dict[str, Any]:

253 r"""Create a new experiment in eLabFTW.

254

255 Parameters

256 ----------

257 title : str

258 Experiment title (required)

259 body : str, optional

260 Experiment body content (supports markdown)

261 tags : list of str, optional

262 List of tag strings to apply

263 metadata : dict, optional

264 Experiment metadata. Can be either:

265

266 1. **Flat key-value pairs** (simple):

267 ``{"key": "value", "number": 123}``

268

269 2. **eLabFTW extra_fields schema** (recommended):

270 ``{``

271 `` "extra_fields": {``

272 `` "Field Name": {``

274 `` "value": "field value",``

275 `` "description": "Optional description",``

276 `` "position": 1,``

277 `` "group_id": 1,``

278 `` ...``

279 `` },``

280 `` ...``

281 `` },``

282 `` "elabftw": {``

283 `` "display_main_text": true,``

284 `` "extra_fields_groups": [``

285 `` {"id": 1, "name": "Group Name"},``

286 `` ...``

287 `` ]``

288 `` }``

289 ``}``

290

291 For extra_fields schema details, see:

292 https://doc.elabftw.net/metadata.html#schema-description

293 category : int, optional

294 Category ID (uses eLabFTW default if not specified)

295 status : int, optional

296 Status ID (uses eLabFTW default if not specified)

297 content_type : ContentType, optional

298 Content type for body text (HTML or MARKDOWN).

299 Defaults to eLabFTW's default (HTML) if not specified.

300

301 Returns

302 -------

303 dict

304 Created experiment data including 'id' field

305

306 Raises

307 ------

308 ELabFTWAuthenticationError

309 If API key is invalid

310 ELabFTWError

311 If creation fails

312

313 Examples

314 --------

315 Simple metadata (flat key-value pairs):

316

317 >>> from nexusLIMS.utils.elabftw import ContentType

318 >>> exp = client.create_experiment(

319 ... title="TEM Analysis",

320 ... body="Sample characterization with TEM",

321 ... tags=["microscopy", "analysis"],

322 ... metadata={"instrument": "FEI Titan", "operator": "jsmith"}

323 ... )

324 >>> print(f"Created experiment ID: {exp['id']}")

325

326 With Markdown body:

327

328 >>> exp = client.create_experiment(

329 ... title="TEM Analysis",

330 ... body=(

331 ... "## Sample Analysis\\n\\n- **Sample**: Steel alloy\\n"

332 ... "- **Method**: TEM imaging"

333 ... ),

334 ... content_type=ContentType.MARKDOWN,

335 ... tags=["microscopy"]

336 ... )

337 >>> print(f"Created experiment ID: {exp['id']}")

338

339 Structured extra_fields (recommended):

340

341 >>> exp = client.create_experiment(

342 ... title="TEM Analysis",

343 ... metadata={

344 ... "extra_fields": {

345 ... "Instrument": {

346 ... "type": "text",

347 ... "value": "FEI Titan",

348 ... "description": "Instrument used",

349 ... "position": 1,

350 ... "group_id": 1

351 ... },

352 ... "Start Time": {

353 ... "type": "datetime-local",

354 ... "value": "2025-01-27T10:30",

355 ... "description": "Session start time",

356 ... "position": 2,

357 ... "group_id": 1

358 ... },

359 ... "CDCS Record": {

360 ... "type": "url",

361 ... "value": "https://cdcs.example.com/record/123",

362 ... "description": "Link to related CDCS record",

363 ... "position": 3,

364 ... "group_id": 2

365 ... }

366 ... },

367 ... "elabftw": {

368 ... "display_main_text": True,

369 ... "extra_fields_groups": [

370 ... {"id": 1, "name": "Session Information"},

371 ... {"id": 2, "name": "Related Records"}

372 ... ]

373 ... }

374 ... }

375 ... )

376 """

377 payload: dict[str, Any] = {"title": title}

378

379 if body is not None:

380 payload["body"] = body

381

382 if tags:

383 # eLabFTW expects list of strings

384 payload["tags"] = tags

385

386 if metadata:

387 payload["metadata"] = metadata

388

389 if category is not None:

390 payload["category"] = category

391

392 if status is not None:

393 payload["status"] = status

394

395 if content_type is not None:

396 payload["content_type"] = int(content_type)

397

398 _logger.debug("Creating experiment: %s", title)

399 result = self._make_request(

400 "POST", self.experiments_endpoint, json_data=payload

401 )

402 _logger.info("Created experiment %s: %s", result["id"], title)

403

404 return result

405

406 def get_experiment(self, experiment_id: int) -> dict[str, Any]:

407 """Retrieve an experiment by ID.

408

409 Parameters

410 ----------

411 experiment_id : int

412 Experiment ID to retrieve

413

414 Returns

415 -------

416 dict

417 Full experiment data. Note that the 'tags' field is automatically

418 converted from eLabFTW's pipe-separated string format to a list of

419 strings for convenience.

420

421 Raises

422 ------

423 ELabFTWNotFoundError

424 If experiment doesn't exist

425 ELabFTWError

426 If retrieval fails

427

428 Examples

429 --------

430 >>> exp = client.get_experiment(42)

431 >>> print(exp['title'])

432 >>> print(exp['tags']) # ['tag1', 'tag2', 'tag3']

433 """

434 url = f"{self.experiments_endpoint}/{experiment_id}"

435 _logger.debug("Fetching experiment %s", experiment_id)

436

437 result = self._make_request("GET", url)

438

439 # _make_request can return None for 204 responses, which shouldn't

440 # happen for GET requests but we handle it defensively

441 if result is None:

442 msg = f"Unexpected empty response when fetching experiment {experiment_id}"

443 raise ELabFTWError(msg)

444

445 # Convert tags from pipe-separated string to list

446 # eLabFTW returns tags as "tag1|tag2|tag3" but we normalize to a list

447 if "tags" in result and isinstance(result["tags"], str):

448 # Empty string should become empty list

449 result["tags"] = result["tags"].split("|") if result["tags"] else []

450

451 # Parse metadata if it's a JSON string

452 # eLabFTW may return metadata as a JSON string, normalize to dict

453 if "metadata" in result and isinstance(result["metadata"], str):

454 try:

455 result["metadata"] = json.loads(result["metadata"])

456 except json.JSONDecodeError:

457 # If parsing fails, leave as string

458 _logger.warning(

459 "Failed to parse metadata JSON for experiment %s", experiment_id

460 )

461

462 return result

463

464 def list_experiments(

465 self,

466 limit: int = 15,

467 offset: int = 0,

468 query: str | None = None,

469 ) -> list[dict[str, Any]]:

470 """List experiments with pagination and optional search.

471

472 Parameters

473 ----------

474 limit : int, optional

475 Maximum number of results to return (default: 15)

476 offset : int, optional

477 Number of results to skip (for pagination) (default: 0)

478 query : str, optional

479 Full-text search query

480

481 Returns

482 -------

483 list of dict

484 List of experiment data dicts

485

486 Raises

487 ------

488 ELabFTWError

489 If listing fails

490

491 Examples

492 --------

493 >>> # Get first 10 experiments

494 >>> experiments = client.list_experiments(limit=10)

495 >>>

496 >>> # Search for experiments

497 >>> results = client.list_experiments(query="microscopy")

498 >>>

499 >>> # Pagination

500 >>> page2 = client.list_experiments(limit=10, offset=10)

501 """

502 params = {"limit": limit, "offset": offset}

503

504 if query:

505 params["q"] = query

506

507 # Build URL with query parameters

508 param_str = "&".join(f"{k}={v}" for k, v in params.items())

509 url = f"{self.experiments_endpoint}?{param_str}"

510

511 _logger.debug(

512 "Listing experiments (limit=%s, offset=%s, query=%s)",

513 limit,

514 offset,

515 query,

516 )

517

518 return self._make_request("GET", url)

519

520 def update_experiment( # noqa: PLR0913

521 self,

522 experiment_id: int,

523 title: str | None = None,

524 body: str | None = None,

525 tags: list[str] | None = None,

526 metadata: dict[str, Any] | None = None,

527 category: int | None = None,

528 status: int | None = None,

529 ) -> dict[str, Any]:

530 """Update an existing experiment.

531

532 Only fields provided as arguments will be updated. Other fields

533 remain unchanged.

534

535 Parameters

536 ----------

537 experiment_id : int

538 ID of experiment to update

539 title : str, optional

540 New title

541 body : str, optional

542 New body content

543 tags : list of str, optional

544 New tag list (replaces existing tags)

545 metadata : dict, optional

546 New metadata (replaces existing metadata). Can be either:

547

548 1. **Flat key-value pairs** (simple):

549 ``{"key": "value", "number": 123}``

550

551 2. **eLabFTW extra_fields schema** (recommended):

552 See ``create_experiment()`` for full schema documentation.

553

554 For extra_fields schema details, see:

555 https://doc.elabftw.net/metadata.html#schema-description

556 category : int, optional

557 New category ID

558 status : int, optional

559 New status ID

560

561 Returns

562 -------

563 dict

564 Updated experiment data

565

566 Raises

567 ------

568 ELabFTWNotFoundError

569 If experiment doesn't exist

570 ELabFTWError

571 If update fails

572

573 Examples

574 --------

575 Update title only:

576

577 >>> client.update_experiment(42, title="New Title")

578

579 Update multiple fields with flat metadata:

580

581 >>> client.update_experiment(

582 ... 42,

583 ... body="Updated description",

584 ... tags=["new-tag"],

585 ... metadata={"updated": "2025-01-31"}

586 ... )

587

588 Update with extra_fields schema:

589

590 >>> client.update_experiment(

591 ... 42,

592 ... metadata={

593 ... "extra_fields": {

594 ... "Status": {

595 ... "type": "text",

596 ... "value": "Completed",

597 ... "description": "Experiment status",

598 ... "position": 1

599 ... }

600 ... },

601 ... "elabftw": {"display_main_text": True}

602 ... }

603 ... )

604 """

605 url = f"{self.experiments_endpoint}/{experiment_id}"

606 payload: dict[str, Any] = {}

607

608 if title is not None:

609 payload["title"] = title

610

611 if body is not None:

612 payload["body"] = body

613

614 if tags is not None:

615 payload["tags"] = tags

616

617 if metadata is not None:

618 payload["metadata"] = metadata

619

620 if category is not None:

621 payload["category"] = category

622

623 if status is not None:

624 payload["status"] = status

625

626 _logger.debug("Updating experiment %s", experiment_id)

627 result = self._make_request("PATCH", url, json_data=payload)

628 _logger.info("Updated experiment %s", experiment_id)

629

630 return result

631

632 def delete_experiment(self, experiment_id: int) -> None:

633 """Delete an experiment.

634

635 Note: This is a soft delete in eLabFTW - the experiment is marked

636 as deleted but can be restored by administrators.

637

638 Parameters

639 ----------

640 experiment_id : int

641 ID of experiment to delete

642

643 Raises

644 ------

645 ELabFTWNotFoundError

646 If experiment doesn't exist

647 ELabFTWError

648 If deletion fails

649

650 Examples

651 --------

652 >>> client.delete_experiment(42)

653 """

654 url = f"{self.experiments_endpoint}/{experiment_id}"

655 _logger.debug("Deleting experiment %s", experiment_id)

656

657 self._make_request("DELETE", url)

658 _logger.info("Deleted experiment %s", experiment_id)

659

660 def upload_file_to_experiment(

661 self,

662 experiment_id: int,

663 file_path: Path | str,

664 comment: str | None = None,

665 ) -> dict[str, Any]:

666 """Upload a file as attachment to an experiment.

667

668 Parameters

669 ----------

670 experiment_id : int

671 ID of experiment to attach file to

672 file_path : pathlib.Path or str

673 Path to file to upload

674 comment : str, optional

675 Comment/description for the uploaded file

676

677 Returns

678 -------

679 dict

680 Upload result data

681

682 Raises

683 ------

684 FileNotFoundError

685 If file doesn't exist

686 ELabFTWNotFoundError

687 If experiment doesn't exist

688 ELabFTWError

689 If upload fails

690

691 Examples

692 --------

693 >>> result = client.upload_file_to_experiment(

694 ... experiment_id=42,

695 ... file_path="data.xml",

696 ... comment="NexusLIMS XML record"

697 ... )

698 """

699 file_path = Path(file_path)

700

701 if not file_path.exists():

702 msg = f"File not found: {file_path}"

703 raise FileNotFoundError(msg)

704

705 url = f"{self.experiments_endpoint}/{experiment_id}/uploads"

706

707 # Prepare multipart/form-data upload

708 with file_path.open("rb") as f:

709 files = {"file": (file_path.name, f, "application/octet-stream")}

710

711 # Add comment as form data if provided

712 data = {"comment": comment} if comment else None

713

714 _logger.debug(

715 "Uploading %s to experiment %s", file_path.name, experiment_id

716 )

717

718 # Note: When files are provided, nexus_req sends as multipart/form-data

719 # We need to pass data as form data, not json

720 response = nexus_req(

721 url,

722 "POST",

723 headers={"Authorization": self.api_key},

724 data=data,

725 files=files,

726 )

727

728 if response.status_code == HTTPStatus.CREATED:

729 _logger.info(

730 "Uploaded %s to experiment %s", file_path.name, experiment_id

731 )

732 # eLabFTW returns Location header with upload URL

733 location = response.headers.get("Location")

734 if location:

735 # Extract upload ID from URL

736 try:

737 upload_id = int(location.rstrip("/").rsplit("/", 1)[-1])

738 except ValueError as e:

739 msg = (

740 "Failed to parse upload ID from "

741 f"Location header: {location}"

742 )

743 raise ELabFTWError(msg) from e

744 else:

745 return {"id": upload_id, "location": location}

746 # Fallback: try to parse JSON response

747 try:

748 return response.json()

749 except Exception:

750 msg = "201 Created response missing Location header and JSON body"

751 raise ELabFTWError(msg) from None

752

753 if response.status_code == HTTPStatus.UNAUTHORIZED:

754 msg = "Authentication failed - check API key"

755 raise ELabFTWAuthenticationError(msg)

756

757 if response.status_code == HTTPStatus.NOT_FOUND:

758 msg = f"Experiment {experiment_id} not found"

759 raise ELabFTWNotFoundError(msg)

760

761 msg = (

762 f"File upload failed with status "

763 f"{response.status_code}: {response.text}"

764 )

765 raise ELabFTWError(msg)

766

767

768def get_elabftw_client() -> ELabFTWClient:

769 """Get configured eLabFTW client from settings.

770

771 Convenience function that creates a client using credentials from

772 the NexusLIMS configuration.

773

774 Returns

775 -------

776 ELabFTWClient

777 Configured client instance

778

779 Raises

780 ------

781 ValueError

782 If NX_ELABFTW_API_KEY or NX_ELABFTW_URL not configured

783

784 Examples

785 --------

786 >>> from nexusLIMS.utils.elabftw import get_elabftw_client

787 >>> client = get_elabftw_client()

788 >>> experiments = client.list_experiments(limit=10)

789 """

790 if not settings.NX_ELABFTW_API_KEY:

791 msg = "NX_ELABFTW_API_KEY not configured"

792 raise ValueError(msg)

793

794 if not settings.NX_ELABFTW_URL:

795 msg = "NX_ELABFTW_URL not configured"

796 raise ValueError(msg)

797

798 return ELabFTWClient(

799 base_url=str(settings.NX_ELABFTW_URL),

800 api_key=settings.NX_ELABFTW_API_KEY,

801 )