Coverage for nexusLIMS/config.py: 100%

1""" 

2Centralized environment variable management for NexusLIMS. 

3 

4This module uses `pydantic-settings` to define, validate, and access 

5application settings from environment variables and .env files. 

6It provides a single source of truth for configuration, ensuring 

7type safety and simplifying access throughout the application. 

8 

9The settings object can be imported and used throughout the application. 

10In tests, use refresh_settings() to reload settings after modifying 

11environment variables. 

12""" 

13 

14import logging 

15import os 

16import re 

17from pathlib import Path 

18from typing import TYPE_CHECKING, Literal 

19 

20from dotenv import dotenv_values 

21from pydantic import ( 

22 AnyHttpUrl, 

23 BaseModel, 

24 DirectoryPath, 

25 EmailStr, 

26 Field, 

27 FilePath, 

28 ValidationError, 

29 field_validator, 

30) 

31from pydantic_settings import BaseSettings, SettingsConfigDict 

32 

33from nexusLIMS.version import __version__ 

34 

35_logger = logging.getLogger(__name__) 

36 

37# ============================================================================ 

38# TEST MODE: Disable Pydantic validation when running tests 

39# ============================================================================ 

40# Check if we're in test mode BEFORE defining the Settings class 

41# This allows tests to control the environment setup without validation errors 

42TEST_MODE = os.getenv("NX_TEST_MODE", "").lower() in ("true", "1", "yes") 

43 

44if TEST_MODE: 

45 _logger.warning("NX_TEST_MODE enabled - Pydantic validation disabled") 

46 

47# Type aliases that conditionally use strict validation types or plain Path 

48# based on TEST_MODE. When TEST_MODE=True, use Path (no existence validation). 

49# When TEST_MODE=False, use DirectoryPath/FilePath (validates existence). 

50if TYPE_CHECKING: 

51 # For type checkers, always use the strict types for proper type hints 

52 TestAwareDirectoryPath = DirectoryPath 

53 TestAwareFilePath = FilePath 

54 TestAwareHttpUrl = AnyHttpUrl 

55 TestAwareEmailStr = EmailStr 

56else: 

57 # At runtime, conditionally use strict or lenient types 

58 TestAwareDirectoryPath = Path if TEST_MODE else DirectoryPath 

59 TestAwareFilePath = Path if TEST_MODE else FilePath 

60 TestAwareHttpUrl = str if TEST_MODE else AnyHttpUrl 

61 TestAwareEmailStr = str if TEST_MODE else EmailStr 

62 

63 

64class NemoHarvesterConfig(BaseModel): 

65 """Configuration for a single NEMO harvester instance.""" 

66 

67 # Uses TestAwareHttpUrl which is str in test mode, AnyHttpUrl in production 

68 address: TestAwareHttpUrl = Field( # type: ignore[valid-type] 

69 "http://localhost:8080/api/" if TEST_MODE else ..., 

70 description=( 

71 "Full path to the root of the NEMO API, with trailing slash included " 

72 "(e.g., `https://nemo.example.com/api/`)" 

73 ), 

74 json_schema_extra={ 

75 "required": True, 

76 "detail": ( 

77 "The full URL to the NEMO API root endpoint, including the trailing " 

78 "slash. For example: `https://nemo.yourinstitution.edu/api/`. This " 

79 "must point to the API root, not the NEMO web interface itself.\n\n" 

80 "You can verify the address is correct by navigating to it in a " 

81 "browser — a valid NEMO API root returns a JSON object listing " 

82 "available endpoints." 

83 ), 

84 }, 

85 ) 

86 token: str = Field( 

87 "test_nemo_token" if TEST_MODE else ..., 

88 description=( 

89 "Authentication token for the NEMO server. Obtain from the 'detailed " 

90 "administration' page of the NEMO installation." 

91 ), 

92 json_schema_extra={ 

93 "required": True, 

94 "detail": ( 

95 "The API authentication token for this NEMO server instance. To " 

96 "obtain: log in to NEMO as an administrator, navigate to the " 

97 "'Detailed administration' page (typically at " 

98 "`/admin/authtoken/token/`), and locate or create a token for the " 

99 "NexusLIMS service account. The token is a 40-character hex string." 

100 ), 

101 }, 

102 ) 

103 strftime_fmt: str = Field( 

104 "%Y-%m-%dT%H:%M:%S%z", 

105 description=( 

106 "Format string to send datetime values to the NEMO API. Uses Python " 

107 "strftime syntax. Default is ISO 8601 format." 

108 ), 

109 json_schema_extra={ 

110 "detail": ( 

111 "The Python strftime format string used when sending datetime values " 

112 "to this NEMO API instance. The default `%Y-%m-%dT%H:%M:%S%z` is " 

113 "ISO 8601 and works with all standard NEMO installations.\n\n" 

114 "Only change this if your NEMO server has a non-standard date format " 

115 "configuration. See https://docs.python.org/3/library/datetime.html" 

116 "#strftime-and-strptime-format-codes for format codes." 

117 ) 

118 }, 

119 ) 

120 strptime_fmt: str = Field( 

121 "%Y-%m-%dT%H:%M:%S%z", 

122 description=( 

123 "Format string to parse datetime values from the NEMO API. Uses Python " 

124 "strptime syntax. Default is ISO 8601 format." 

125 ), 

126 json_schema_extra={ 

127 "detail": ( 

128 "The Python strptime format string used when parsing datetime values " 

129 "returned by this NEMO API instance. The default " 

130 "`%Y-%m-%dT%H:%M:%S%z` " 

131 "is ISO 8601 and works with all standard NEMO installations.\n\n" 

132 "Only change this if your NEMO server returns dates in a non-standard " 

133 "format. See https://docs.python.org/3/library/datetime.html" 

134 "#strftime-and-strptime-format-codes for format codes." 

135 ) 

136 }, 

137 ) 

138 tz: str | None = Field( 

139 None, 

140 description=( 

141 "IANA timezone name (e.g., `America/Denver`) to coerce API datetime " 

142 "strings into. Only needed if the NEMO server doesn't return timezone " 

143 "information in API responses. If provided, overrides timezone from API." 

144 ), 

145 json_schema_extra={ 

146 "detail": ( 

147 "An IANA tz database timezone name (e.g., `America/Denver`, " 

148 "`Europe/Berlin`) to force onto datetime values received from this " 

149 "NEMO server. Only needed when your NEMO server returns " 

150 "reservation/usage event times without timezone information.\n\n" 

151 "Leave blank for NEMO servers that include timezone info in their " 

152 "API responses. See " 

153 "https://en.wikipedia.org/wiki/List_of_tz_database_time_zones " 

154 "for valid timezone names." 

155 ) 

156 }, 

157 ) 

158 

159 @field_validator("address") 

160 @classmethod 

161 def validate_trailing_slash(cls, v: str | AnyHttpUrl) -> str | AnyHttpUrl: 

162 """Ensure the API address has a trailing slash.""" 

163 if TEST_MODE: 

164 return v # Skip validation in test mode 

165 if not str(v).endswith("/"): 

166 msg = "NEMO address must end with a trailing slash" 

167 raise ValueError(msg) 

168 return v 

169 

170 

171class EmailConfig(BaseModel): 

172 """Config for email notifications from the nexuslims build-records script.""" 

173 

174 smtp_host: str = Field( 

175 "localhost" if TEST_MODE else ..., 

176 description="SMTP server hostname (e.g., 'smtp.gmail.com')", 

177 json_schema_extra={ 

178 "required": True, 

179 "detail": ( 

180 "The hostname or IP address of the SMTP server used to send " 

181 "error notification emails. For Gmail use `smtp.gmail.com`, for " 

182 "Outlook/Office365 use `smtp.office365.com`. For an on-premises " 

183 "mail relay this is typically a local hostname or IP address." 

184 ), 

185 }, 

186 ) 

187 smtp_port: int = Field( 

188 587, 

189 description="SMTP server port (default: 587 for STARTTLS)", 

190 json_schema_extra={ 

191 "detail": ( 

192 "The TCP port for the SMTP connection. Common values:\n" 

193 " `587` — STARTTLS (recommended, default)\n" 

194 " `465` — SMTPS / implicit TLS\n" 

195 " `25` — unencrypted (not recommended)\n\n" 

196 "The default `587` works with most modern mail servers when " 

197 "Use TLS is enabled." 

198 ) 

199 }, 

200 ) 

201 smtp_username: str | None = Field( 

202 None, 

203 description="SMTP username for authentication (if required)", 

204 json_schema_extra={ 

205 "detail": ( 

206 "The username for SMTP authentication. For Gmail this is your " 

207 "full email address. Leave blank if your SMTP relay does not " 

208 "require authentication (e.g., an internal relay that accepts " 

209 "connections from trusted hosts without credentials)." 

210 ) 

211 }, 

212 ) 

213 smtp_password: str | None = Field( 

214 None, 

215 description="SMTP password for authentication (if required)", 

216 json_schema_extra={ 

217 "detail": ( 

218 "The password for SMTP authentication. For Gmail, use an App " 

219 "Password (not your account password) if 2-factor authentication " 

220 "is enabled. Leave blank if your SMTP relay does not require " 

221 "authentication." 

222 ) 

223 }, 

224 ) 

225 use_tls: bool = Field( 

226 default=True, 

227 description="Use TLS encryption (default: True)", 

228 json_schema_extra={ 

229 "detail": ( 

230 "Whether to use TLS encryption for the SMTP connection. When " 

231 "enabled (the default), NexusLIMS uses STARTTLS on the configured " 

232 "port (typically 587). Disable only when connecting to a plaintext " 

233 "SMTP relay on port 25." 

234 ) 

235 }, 

236 ) 

237 sender: TestAwareEmailStr = Field( # type: ignore[valid-type] 

238 "test@example.com" if TEST_MODE else ..., 

239 description="Email address to send from", 

240 json_schema_extra={ 

241 "required": True, 

242 "detail": ( 

243 "The 'From' email address for notification messages. This must " 

244 "be an address that your SMTP server is authorized to send from. " 

245 "If using Gmail or similar services, this must match the " 

246 "authenticated account's address." 

247 ), 

248 }, 

249 ) 

250 recipients: list[TestAwareEmailStr] = Field( # type: ignore[valid-type] 

251 ["test@example.com"] if TEST_MODE else ..., 

252 description="List of recipient email addresses for error notifications", 

253 json_schema_extra={ 

254 "required": True, 

255 "detail": ( 

256 "One or more email addresses that will receive error notification " 

257 "messages when the record builder encounters problems. Provide as " 

258 "a comma-separated string, e.g.:\n" 

259 " `NX_EMAIL_RECIPIENTS='admin@example.com,team@example.com'`\n\n" 

260 "Notifications are sent when `nexuslims build-records` detects " 

261 "ERROR-level log entries." 

262 ), 

263 }, 

264 ) 

265 

266 

267class Settings(BaseSettings): 

268 """ 

269 Manage application settings loaded from environment variables and `.env` files. 

270 

271 This class utilizes `pydantic-settings` to provide a robust and type-safe way 

272 to define, validate, and access all application configurations. It ensures 

273 that settings are loaded with proper types and provides descriptions for each. 

274 """ 

275 

276 model_config = SettingsConfigDict( 

277 # CRITICAL: Disable .env file loading in TEST_MODE to prevent contamination 

278 # from local environment files during testing 

279 env_file=None if TEST_MODE else ".env", 

280 env_file_encoding="utf-8", 

281 extra="ignore", # Ignore extra environment variables not defined here 

282 # In test mode, disable path validation to allow non-existent paths 

283 validate_default=not TEST_MODE, 

284 ) 

285 

286 NX_FILE_STRATEGY: Literal["exclusive", "inclusive"] = Field( 

287 "exclusive", 

288 description=( 

289 "Defines how file finding will behave: 'exclusive' (only files with " 

290 "explicit extractors) or 'inclusive' (all files, with basic metadata " 

291 "for others). Default is 'exclusive'." 

292 ), 

293 json_schema_extra={ 

294 "detail": ( 

295 "Controls which files are included when searching for experiment " 

296 "data.\n\n" 

297 "`exclusive` (default): Only files for which NexusLIMS has an explicit " 

298 "metadata extractor are included. This produces cleaner records but " 

299 "may miss ancillary files.\n\n" 

300 "`inclusive`: All files within the session window are included. Files " 

301 "without a known extractor receive basic filesystem metadata only. " 

302 "Useful when you want a complete audit trail of every file created " 

303 "during an instrument session.\n\n" 

304 "See https://datasophos.github.io/NexusLIMS/stable/" 

305 "user_guide/extractors.html " 

306 "for the list of supported file formats and their extractors." 

307 ) 

308 }, 

309 ) 

310 NX_IGNORE_PATTERNS: list[str] = Field( 

311 ["*.mib", "*.db", "*.emi", "*.hdr"], 

312 description=( 

313 "List of glob patterns to ignore when searching for experiment files. " 

314 "Default is `['*.mib','*.db','*.emi','*.hdr']`." 

315 ), 

316 json_schema_extra={ 

317 "detail": ( 

318 "Filename glob patterns to exclude when scanning for experiment files. " 

319 "Patterns follow the same syntax as the '-name' argument to GNU find " 

320 "(see https://manpages.org/find).\n\n" 

321 "This is stored in the config file as a JSON array string:\n" 

322 ' `NX_IGNORE_PATTERNS=\'["*.mib","*.db","*.emi","*.hdr"]\'`\n\n' 

323 "In the config editor, enter patterns as a comma-separated list.\n\n" 

324 "Common patterns to ignore:\n" 

325 " `*.mib` — Merlin detector raw frames (very large)\n" 

326 " `*.db` — SQLite lock/temp files\n" 

327 " `*.emi` — FEI TIA sidecar files (paired with .ser via FEI EMI " 

328 "extractor)\n" 

329 " `*.hdr` — header files paired with other data formats" 

330 ) 

331 }, 

332 ) 

333 # Use TestAware types which are strict in production, lenient in test mode 

334 NX_INSTRUMENT_DATA_PATH: TestAwareDirectoryPath = Field( # type: ignore[valid-type] 

335 Path("/tmp") / "test_instrument_data" if TEST_MODE else ..., # noqa: S108 

336 description=( 

337 "Root path to the centralized file store for instrument data " 

338 "(mounted read-only). The directory must exist." 

339 ), 

340 json_schema_extra={ 

341 "required": True, 

342 "detail": ( 

343 "The root path to the centralized instrument data file store — " 

344 "typically a network share or mounted volume containing subdirectories " 

345 "for each instrument.\n\n" 

346 "IMPORTANT: This path should be mounted read-only to ensure data " 

347 "preservation. NexusLIMS will never write to this location.\n\n" 

348 "The `filestore_path` column in the NexusLIMS instruments database " 

349 "stores paths relative to this root. For example, if an instrument " 

350 "has `filestore_path='FEI_Titan/data'` and this value is " 

351 "`/mnt/instrument_data`, NexusLIMS searches under " 

352 "`/mnt/instrument_data/FEI_Titan/data`." 

353 ), 

354 }, 

355 ) 

356 NX_DATA_PATH: TestAwareDirectoryPath = Field( # type: ignore[valid-type] 

357 Path("/tmp") / "test_data" if TEST_MODE else ..., # noqa: S108 

358 description=( 

359 "Writable path parallel to NX_INSTRUMENT_DATA_PATH for " 

360 "extracted metadata and generated preview images. The directory must exist." 

361 ), 

362 json_schema_extra={ 

363 "required": True, 

364 "detail": ( 

365 "A writable path that mirrors the directory structure of " 

366 "`NX_INSTRUMENT_DATA_PATH`. NexusLIMS writes extracted metadata files " 

367 "and generated preview images here, alongside the original data.\n\n" 

368 "This path must be accessible to the NexusLIMS CDCS frontend instance " 

369 "so it can serve preview images and metadata files to users browsing " 

370 "records. Configure your CDCS deployment to mount or serve files from " 

371 "this location." 

372 ), 

373 }, 

374 ) 

375 NX_DB_PATH: TestAwareFilePath = Field( # type: ignore[valid-type] 

376 Path("/tmp") / "test.db" if TEST_MODE else ..., # noqa: S108 

377 description=( 

378 "The writable path to the NexusLIMS SQLite database that is used to get " 

379 "information about instruments and sessions that are built into records." 

380 ), 

381 json_schema_extra={ 

382 "required": True, 

383 "detail": ( 

384 "The full filesystem path to the NexusLIMS SQLite database file. " 

385 "\n\n" 

386 "Must be writable by the NexusLIMS process. The database is created " 

387 "automatically on first run of `nexuslims db init`. Recommended " 

388 "location: within `NX_DATA_PATH` for co-location with other data." 

389 ), 

390 }, 

391 ) 

392 NX_CDCS_TOKEN: str = Field( 

393 "test_token" if TEST_MODE else ..., 

394 description=( 

395 "API token for authenticating to the CDCS API for uploading " 

396 "built records to the NexusLIMS front-end." 

397 ), 

398 json_schema_extra={ 

399 "required": True, 

400 "detail": ( 

401 "The API authentication token for the NexusLIMS CDCS frontend. " 

402 "Used for all record upload requests.\n\n" 

403 "To obtain: log in to your CDCS instance as an administrator, navigate " 

404 "to the admin panel, and find or create an API token for the NexusLIMS " 

405 "service account. Alternatively, use the CDCS REST API token " 

406 "endpoint.\n\n" 

407 "Keep this value secret — anyone with this token can upload records " 

408 "to your CDCS instance." 

409 ), 

410 }, 

411 ) 

412 NX_CDCS_URL: TestAwareHttpUrl = Field( # type: ignore[valid-type] 

413 "http://localhost:8000" if TEST_MODE else ..., 

414 description=( 

415 "The root URL of the NexusLIMS CDCS front-end. This will be the target for " 

416 "record uploads that are authenticated using the CDCS credentials." 

417 ), 

418 json_schema_extra={ 

419 "required": True, 

420 "detail": ( 

421 "The root URL of the NexusLIMS CDCS frontend instance. All record " 

422 "uploads are sent here using `NX_CDCS_TOKEN`.\n\n" 

423 "Include the trailing slash: `https://nexuslims.example.com/`\n\n" 

424 "This is the same URL users visit to browse experiment records. " 

425 "NexusLIMS POSTs new XML records to the CDCS REST API at this address." 

426 ), 

427 }, 

428 ) 

429 NX_LABARCHIVES_ACCESS_KEY_ID: str | None = Field( 

430 "test_la_akid" if TEST_MODE else None, 

431 description=( 

432 "Access Key ID (akid) assigned by LabArchives for API access. " 

433 "Required for LabArchives export. If not configured, LabArchives " 

434 "export will be disabled." 

435 ), 

436 json_schema_extra={ 

437 "detail": ( 

438 "The Access Key ID (akid) provided by LabArchives for your API " 

439 "integration. This is the public identifier used alongside " 

440 "`NX_LABARCHIVES_ACCESS_PASSWORD` to sign API requests with " 

441 "HMAC-SHA-512.\n\n" 

442 "Obtain this from your LabArchives account under API settings. " 

443 "Both the Access Key ID and Access Password are required for " 

444 "LabArchives export to be enabled." 

445 ) 

446 }, 

447 ) 

448 NX_LABARCHIVES_ACCESS_PASSWORD: str | None = Field( 

449 "test_la_password" if TEST_MODE else None, 

450 description=( 

451 "Access password (provided by LabArchives) for the API. " 

452 "Used to generate HMAC-SHA-512 signatures for authenticated requests. " 

453 "If not configured, LabArchives export will be disabled." 

454 ), 

455 json_schema_extra={ 

456 "display_default": None, 

457 "detail": ( 

458 "The access password provided by LabArchives for your API " 

459 "integration. This secret is used to generate HMAC-SHA-512 " 

460 "signatures that authenticate all API requests.\n\n" 

461 "Keep this value secret — it is equivalent to a password and " 

462 "allows full API access to the configured LabArchives account. " 

463 "Used together with `NX_LABARCHIVES_ACCESS_KEY_ID`." 

464 ), 

465 }, 

466 ) 

467 NX_LABARCHIVES_USER_ID: str | None = Field( 

468 "test_la_uid" if TEST_MODE else None, 

469 description=( 

470 "LabArchives user ID (uid) for the account that owns records. " 

471 "Obtain once via the `user_access_info` API call. " 

472 "Required for LabArchives export." 

473 ), 

474 json_schema_extra={ 

475 "detail": ( 

476 "The persistent user ID (uid) for the LabArchives account that " 

477 "will own uploaded notebook entries. This is obtained once via " 

478 "the LabArchives `user_access_info` API call.\n\n" 

479 "To get the uid: call the LabArchives API endpoint " 

480 "`users/user_access_info` with your login and token. " 

481 "The returned `uid` value is what you need here.\n\n" 

482 "This uid is stable and does not change, so you only need to " 

483 "retrieve it once." 

484 ) 

485 }, 

486 ) 

487 NX_LABARCHIVES_URL: TestAwareHttpUrl | None = Field( # type: ignore[valid-type] 

488 "http://localhost:9000" if TEST_MODE else "https://api.labarchives.com/api", 

489 description=( 

490 "Base URL for the LabArchives API (including the /api path). " 

491 "Defaults to https://api.labarchives.com/api for cloud LabArchives. " 

492 "If not configured, LabArchives export will be disabled." 

493 ), 

494 json_schema_extra={ 

495 "detail": ( 

496 "The base URL for LabArchives API calls, including the `/api` path " 

497 "segment. For the standard cloud service this is " 

498 "`https://api.labarchives.com/api`. For self-hosted instances use " 

499 "`https://your-instance.example.com/api`.\n\n" 

500 "All API requests are sent to `{NX_LABARCHIVES_URL}/{class}/{method}`, " 

501 "so the `/api` suffix must be included here." 

502 ) 

503 }, 

504 ) 

505 NX_LABARCHIVES_NOTEBOOK_ID: str | None = Field( 

506 None, 

507 description=( 

508 "Target notebook ID (nbid) for LabArchives uploads. If not set, " 

509 "records are uploaded to the user's Inbox notebook." 

510 ), 

511 json_schema_extra={ 

512 "detail": ( 

513 "The notebook ID (nbid) of the target LabArchives notebook where " 

514 "NexusLIMS records will be stored. When configured, NexusLIMS " 

515 "automatically creates a `NexusLIMS Records/{instrument}/` folder " 

516 "hierarchy within this notebook.\n\n" 

517 "If not set, records are uploaded to the user's default Inbox.\n\n" 

518 "To find the nbid: open the notebook in LabArchives and check " 

519 "the URL — it typically appears as a parameter in the notebook URL." 

520 ) 

521 }, 

522 ) 

523 NX_EXPORT_STRATEGY: Literal["all", "first_success", "best_effort"] = Field( 

524 "all", 

525 description=( 

526 "Strategy for exporting records to multiple destinations. " 

527 "'all': All destinations must succeed (recommended). " 

528 "'first_success': Stop after first successful export. " 

529 "'best_effort': Try all destinations, succeed if any succeed." 

530 ), 

531 json_schema_extra={ 

532 "detail": ( 

533 "Controls behavior when exporting records to multiple destinations " 

534 "(e.g., both CDCS and eLabFTW are configured):\n\n" 

535 "`all` (default, recommended): Every configured destination must " 

536 "accept the record. If any destination fails, the session is marked " 

537 "`ERROR` and retried on the next run.\n\n" 

538 "`first_success`: Stop after the first destination that accepts the " 

539 "record. Useful if destinations are fallbacks for each other.\n\n" 

540 "`best_effort`: Attempt all destinations; mark `COMPLETED` if at least " 

541 "one succeeds. Failed destinations are logged but do not trigger " 

542 "a retry." 

543 ) 

544 }, 

545 ) 

546 NX_CERT_BUNDLE_FILE: TestAwareFilePath | None = Field( 

547 None, 

548 description=( 

549 "If needed, a custom SSL certificate CA bundle can be used to verify " 

550 "requests to the CDCS or NEMO APIs. Provide this value with the full " 

551 "path to a certificate bundle and any certificates provided in the " 

552 "bundle will be appended to the existing system for all requests made " 

553 "by NexusLIMS." 

554 ), 

555 json_schema_extra={ 

556 "detail": ( 

557 "Path to a custom SSL/TLS CA bundle file in PEM format. Use this " 

558 "when your CDCS or NEMO servers use certificates signed by a private " 

559 "or institutional CA not in the system trust store.\n\n" 

560 "Any certificates in this bundle are appended to the existing system " 

561 "CA certificates — they do not replace them. Provide the full absolute " 

562 "path to the `.pem` or `.crt` file.\n\n" 

563 "If both `NX_CERT_BUNDLE` and `NX_CERT_BUNDLE_FILE` are set, " 

564 "`NX_CERT_BUNDLE` takes precedence." 

565 ) 

566 }, 

567 ) 

568 NX_CERT_BUNDLE: str | None = Field( 

569 None, 

570 description=( 

571 "As an alternative to NX_CERT_BUNDLE_FILE, to you can provide the entire " 

572 "certificate bundle as a single string (this can be useful for CI/CD " 

573 "pipelines). If defined, this value will take precedence over " 

574 "NX_CERT_BUNDLE_FILE." 

575 ), 

576 json_schema_extra={ 

577 "detail": ( 

578 "The full text of a PEM-format CA certificate bundle, provided " 

579 "directly as a string rather than a file path. Certificate lines " 

580 "should be separated by '\\n' in the .env file, or just " 

581 "pasted into the config editor field.\n\n" 

582 "This is primarily useful in CI/CD pipelines or containerized " 

583 "deployments where injecting a certificate file is impractical but " 

584 "environment variables are easy to set as secrets.\n\n" 

585 "When defined, this value takes precedence over `NX_CERT_BUNDLE_FILE`." 

586 ) 

587 }, 

588 ) 

589 NX_DISABLE_SSL_VERIFY: bool = Field( 

590 default=False, 

591 description=( 

592 "Disable SSL certificate verification for all outgoing HTTPS requests. " 

593 "This should ONLY be used during local development or testing with " 

594 "self-signed certificates. Never enable this in production." 

595 ), 

596 json_schema_extra={ 

597 "detail": ( 

598 "WARNING: Disables SSL certificate verification for ALL outgoing " 

599 "HTTPS requests, including connections to CDCS, NEMO, and eLabFTW.\n\n" 

600 "NEVER enable this in production. An attacker could intercept all " 

601 "communications including API tokens and uploaded records.\n\n" 

602 "Only appropriate for local development or testing with self-signed " 

603 "certificates when setting up a CA via `NX_CERT_BUNDLE_FILE` is " 

604 "impractical. If you need this in production, configure " 

605 "`NX_CERT_BUNDLE_FILE` instead." 

606 ) 

607 }, 

608 ) 

609 NX_FILE_DELAY_DAYS: float = Field( 

610 2.0, 

611 description=( 

612 "NX_FILE_DELAY_DAYS controls the maximum delay between observing a " 

613 "session ending and when the files are expected to be present. For the " 

614 "number of days set below (can be a fraction of a day, if desired), record " 

615 "building will not fail if no files are found, and the builder will search " 

616 'for again until the delay has passed. So if the value is "2", and a ' 

617 "session ended Monday at 5PM, the record builder will continue to try " 

618 "looking for files until Wednesday at 5PM. " 

619 ), 

620 gt=0, 

621 json_schema_extra={ 

622 "detail": ( 

623 "The maximum time (in days) to wait for instrument files to appear " 

624 "after a session ends before giving up.\n\n" 

625 "Background: On some systems, instrument data files are not " 

626 "immediately available on the network share after an experiment ends " 

627 "— they may be synced or transferred with a delay.\n\n" 

628 "When a session ends and no files are found, the record builder marks " 

629 "it `NO_FILES_FOUND` and retries on subsequent runs until this window " 

630 "expires.\n\n" 

631 "Example: With a value of 2, if a session ended Monday at 5 PM, " 

632 "the builder keeps retrying until Wednesday at 5 PM.\n\n" 

633 "Can be a fractional value (e.g., 0.5 for 12 hours). Must be > 0." 

634 ) 

635 }, 

636 ) 

637 NX_CLUSTERING_SENSITIVITY: float = Field( 

638 1.0, 

639 description=( 

640 "Controls the sensitivity of file clustering into Acquisition Activities. " 

641 "Higher values (e.g., 2.0) make clustering more sensitive to time gaps, " 

642 "resulting in more activities. Lower values (e.g., 0.5) make clustering " 

643 "less sensitive, resulting in fewer activities. Set to 0 to disable " 

644 "clustering entirely and group all files into a single activity. " 

645 "Default is 1.0 (no change to automatic clustering)." 

646 ), 

647 ge=0, 

648 json_schema_extra={ 

649 "detail": ( 

650 "Controls how aggressively files are grouped into separate Acquisition " 

651 "Activities within a session record.\n\n" 

652 "NexusLIMS uses kernel density estimation (KDE) on file modification " 

653 "times to detect natural gaps in activity. This multiplier scales the " 

654 "KDE bandwidth.\n\n" 

655 "Higher values (e.g., `2.0`): more sensitive — smaller time gaps cause " 

656 "a split, producing more (smaller) activities.\n\n" 

657 "Lower values (e.g., `0.5`): less sensitive — only large gaps cause a " 

658 "split, producing fewer (larger) activities.\n\n" 

659 "Set to `0` to disable clustering and place all files into a single " 

660 "Acquisition Activity. Default is `1.0` (unmodified KDE bandwidth)." 

661 ) 

662 }, 

663 ) 

664 NX_LOG_PATH: TestAwareDirectoryPath | None = Field( # type: ignore[valid-type] 

665 None, 

666 description=( 

667 "Directory for application logs. If not specified, defaults to " 

668 "NX_DATA_PATH/logs/. Logs are organized by date: logs/YYYY/MM/DD/" 

669 ), 

670 json_schema_extra={ 

671 "detail": ( 

672 "Directory for NexusLIMS application logs. If not specified, logs " 

673 "are written to `NX_DATA_PATH/logs/` by default.\n\n" 

674 "Within this directory, logs are organized by date:\n" 

675 " `YYYY/MM/DD/YYYYMMDD-HHMM.log`\n\n" 

676 "The directory must be writable by the NexusLIMS process. Leave " 

677 "blank to use the default location within `NX_DATA_PATH`." 

678 ) 

679 }, 

680 ) 

681 NX_RECORDS_PATH: TestAwareDirectoryPath | None = Field( 

682 None, 

683 description=( 

684 "Directory for generated XML records. If not specified, defaults to " 

685 "NX_DATA_PATH/records/. Successfully uploaded records are moved to " 

686 "a 'uploaded' subdirectory." 

687 ), 

688 json_schema_extra={ 

689 "detail": ( 

690 "Directory where generated XML record files are stored before and " 

691 "after upload. If not specified, defaults to " 

692 "`NX_DATA_PATH/records/`.\n\n" 

693 "After a record is successfully uploaded, the XML file is moved to " 

694 "an `'uploaded'` subdirectory within this path for archival.\n\n" 

695 "Failed records remain in the main directory for inspection. The " 

696 "directory must be writable by the NexusLIMS process." 

697 ) 

698 }, 

699 ) 

700 NX_LOCAL_PROFILES_PATH: TestAwareDirectoryPath | None = Field( 

701 None, 

702 description=( 

703 "Directory for site-specific instrument profiles. These profiles " 

704 "customize metadata extraction for instruments unique to your deployment " 

705 "without modifying the core NexusLIMS codebase. Profile files should be " 

706 "Python modules that register InstrumentProfile objects. If not specified, " 

707 "only built-in profiles will be loaded." 

708 ), 

709 json_schema_extra={ 

710 "detail": ( 

711 "Directory containing site-specific instrument profile Python modules. " 

712 "Profiles customize metadata extraction for instruments unique to your " 

713 "deployment without modifying the core NexusLIMS codebase.\n\n" 

714 "Each Python file in this directory should define one or more " 

715 "`InstrumentProfile` subclasses that are auto-discovered and loaded " 

716 "alongside built-in profiles.\n\n" 

717 "Use cases: adding static metadata fields, transforming extracted " 

718 "values, adding instrument-specific warnings, or overriding which " 

719 "extractor handles a particular instrument's files.\n\n" 

720 "Leave blank if you only need the built-in profiles." 

721 ) 

722 }, 

723 ) 

724 

725 # ======================================================================== 

726 # eLabFTW Export Destination Configuration (Optional) 

727 # ======================================================================== 

728 NX_ELABFTW_API_KEY: str | None = Field( 

729 "1-" + "a" * 84 if TEST_MODE else None, 

730 description=( 

731 "API key for authenticating to the eLabFTW API. Obtain from the user " 

732 "panel in your eLabFTW instance. If not configured, eLabFTW export will " 

733 "be disabled." 

734 ), 

735 json_schema_extra={ 

736 "display_default": None, 

737 "detail": ( 

738 "API key for authenticating to the eLabFTW API. If not configured, " 

739 "eLabFTW export will be disabled.\n\n" 

740 "To obtain: log in to your eLabFTW instance, go to user settings, " 

741 "and find the 'API keys' section. Create a new key for NexusLIMS.\n\n" 

742 "Format: `{id}-{key}` (e.g., `1-abc123...`). The key is typically " 

743 "a long alphanumeric string." 

744 ), 

745 }, 

746 ) 

747 NX_ELABFTW_URL: TestAwareHttpUrl | None = Field( # type: ignore[valid-type] 

748 "http://elabftw.localhost:40080" if TEST_MODE else None, 

749 description=( 

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

751 "If not configured, eLabFTW export will be disabled." 

752 ), 

753 json_schema_extra={ 

754 "display_default": None, 

755 "detail": ( 

756 "The root URL of your eLabFTW instance. If not configured, eLabFTW " 

757 "export will be disabled.\n\n" 

758 "Should NOT include `/api/` or any path — just the domain:\n" 

759 " `https://elabftw.example.com`\n" 

760 " `http://localhost:3148`\n\n" 

761 "NexusLIMS appends the appropriate API paths automatically." 

762 ), 

763 }, 

764 ) 

765 NX_ELABFTW_EXPERIMENT_CATEGORY: int | None = Field( 

766 None, 

767 description=( 

768 "Default category ID for created experiments. If not specified, " 

769 "eLabFTW will use its default category. Category IDs can be found " 

770 "in the eLabFTW admin panel." 

771 ), 

772 json_schema_extra={ 

773 "detail": ( 

774 "The default category ID for experiments created in eLabFTW. " 

775 "If not specified, eLabFTW uses its default category.\n\n" 

776 "To find category IDs: log in to eLabFTW as an administrator, " 

777 "navigate to the admin panel, and look under 'Experiment categories'. " 

778 "The ID is shown next to each category name." 

779 ) 

780 }, 

781 ) 

782 NX_ELABFTW_EXPERIMENT_STATUS: int | None = Field( 

783 None, 

784 description=( 

785 "Default status ID for created experiments. If not specified, " 

786 "eLabFTW will use its default status. Status IDs can be found " 

787 "in the eLabFTW admin panel." 

788 ), 

789 json_schema_extra={ 

790 "detail": ( 

791 "The default status ID for experiments created in eLabFTW. " 

792 "If not specified, eLabFTW uses its default status.\n\n" 

793 "To find status IDs: log in to eLabFTW as an administrator, " 

794 "navigate to the admin panel, and look under 'Experiment statuses'. " 

795 "The ID is shown next to each status name." 

796 ) 

797 }, 

798 ) 

799 

800 @property 

801 def nexuslims_instrument_data_path(self) -> Path: 

802 """Alias for NX_INSTRUMENT_DATA_PATH for easier access.""" 

803 return self.NX_INSTRUMENT_DATA_PATH 

804 

805 @property 

806 def lock_file_path(self) -> Path: 

807 """Path to the record builder lock file.""" 

808 return self.NX_DATA_PATH / ".builder.lock" 

809 

810 @property 

811 def log_dir_path(self) -> Path: 

812 """Base directory for timestamped log files.""" 

813 return self.NX_LOG_PATH if self.NX_LOG_PATH else self.NX_DATA_PATH / "logs" 

814 

815 @property 

816 def records_dir_path(self) -> Path: 

817 """Base directory for generated XML records.""" 

818 if self.NX_RECORDS_PATH: 

819 return self.NX_RECORDS_PATH 

820 return self.NX_DATA_PATH / "records" 

821 

822 def nemo_harvesters(self) -> dict[int, NemoHarvesterConfig]: 

823 """ 

824 Dynamically discover and parse all NEMO harvester configurations. 

825 

826 Searches environment variables for NX_NEMO_ADDRESS_N patterns and 

827 constructs NemoHarvesterConfig objects for each numbered harvester. 

828 

829 Returns 

830 ------- 

831 dict[int, NemoHarvesterConfig] 

832 Dictionary mapping harvester number (1, 2, 3, ...) to configuration 

833 objects. Empty dict if no harvesters are configured. 

834 

835 Examples 

836 -------- 

837 With environment variables: 

838 

839 ```python 

840 NX_NEMO_ADDRESS_1=https://nemo1.com/api/ 

841 NX_NEMO_TOKEN_1=token123 

842 NX_NEMO_ADDRESS_2=https://nemo2.com/api/ 

843 NX_NEMO_TOKEN_2=token456 

844 NX_NEMO_TZ_2=America/New_York 

845 ``` 

846 

847 The resulting output will be of the following format: 

848 

849 ```python 

850 { 

851 1: NemoHarvesterConfig( 

852 address='https://nemo1.com/api/', token='token123', ... 

853 ), 

854 2: NemoHarvesterConfig( 

855 address='https://nemo2.com/api/', 

856 token='token456', 

857 tz='America/New_York', 

858 ... 

859 ) 

860 } 

861 ``` 

862 """ 

863 harvesters = {} 

864 

865 # CRITICAL: In TEST_MODE, do NOT load from .env file to prevent 

866 # test contamination from local environment configuration 

867 env_vars = {} 

868 if not TEST_MODE: 

869 # Load .env file to get NEMO variables (Pydantic doesn't load 

870 # variables that aren't defined as fields) 

871 env_file_path = Path(".env") 

872 if env_file_path.exists(): 

873 env_vars = dotenv_values(env_file_path) 

874 

875 # Merge with os.environ (os.environ takes precedence) 

876 all_env = {**env_vars, **os.environ} 

877 

878 # Find all NX_NEMO_ADDRESS_N environment variables 

879 address_pattern = re.compile(r"^NX_NEMO_ADDRESS_(\d+)$") 

880 

881 for env_var in all_env: 

882 match = address_pattern.match(env_var) 

883 if match: 

884 harvester_num = int(match.group(1)) 

885 

886 # Get required address and token 

887 address = all_env.get(f"NX_NEMO_ADDRESS_{harvester_num}") 

888 token = all_env.get(f"NX_NEMO_TOKEN_{harvester_num}") 

889 

890 if not address or not token: 

891 _logger.warning( 

892 "Skipping NEMO harvester %d: " 

893 "both NX_NEMO_ADDRESS_%d and " 

894 "NX_NEMO_TOKEN_%d must be set", 

895 harvester_num, 

896 harvester_num, 

897 harvester_num, 

898 ) 

899 continue 

900 

901 # Build config dict with optional fields 

902 config_dict = { 

903 "address": address, 

904 "token": token, 

905 } 

906 

907 # Add optional fields if present 

908 if strftime_fmt := all_env.get(f"NX_NEMO_STRFTIME_FMT_{harvester_num}"): 

909 config_dict["strftime_fmt"] = strftime_fmt 

910 

911 if strptime_fmt := all_env.get(f"NX_NEMO_STRPTIME_FMT_{harvester_num}"): 

912 config_dict["strptime_fmt"] = strptime_fmt 

913 

914 if tz := all_env.get(f"NX_NEMO_TZ_{harvester_num}"): 

915 config_dict["tz"] = tz 

916 

917 try: 

918 harvesters[harvester_num] = NemoHarvesterConfig(**config_dict) 

919 except ValidationError: 

920 _logger.exception( 

921 "Invalid configuration for NEMO harvester %d", 

922 harvester_num, 

923 ) 

924 raise 

925 

926 return harvesters 

927 

928 def email_config(self) -> EmailConfig | None: 

929 """ 

930 Load email configuration from environment variables if available. 

931 

932 This method is cached per Settings instance for performance. 

933 

934 Returns 

935 ------- 

936 EmailConfig | None 

937 Email configuration object if all required settings are present, 

938 None otherwise (email notifications will be disabled). 

939 

940 Examples 

941 -------- 

942 With environment variables: 

943 

944 ```python 

945 NX_EMAIL_SMTP_HOST=smtp.gmail.com 

946 NX_EMAIL_SENDER=nexuslims@example.com 

947 NX_EMAIL_RECIPIENTS=admin@example.com,team@example.com 

948 ``` 

949 

950 Optional variables: 

951 

952 ```python 

953 NX_EMAIL_SMTP_PORT=587 

954 NX_EMAIL_SMTP_USERNAME=user@example.com 

955 NX_EMAIL_SMTP_PASSWORD=secret 

956 NX_EMAIL_USE_TLS=true 

957 ``` 

958 """ 

959 # CRITICAL: In TEST_MODE, do NOT load from .env file to prevent 

960 # test contamination from local environment configuration 

961 env_vars = {} 

962 if not TEST_MODE: 

963 # Load .env file to get email variables 

964 env_file_path = Path(".env") 

965 if env_file_path.exists(): 

966 env_vars = dotenv_values(env_file_path) 

967 

968 # Merge with os.environ (os.environ takes precedence) 

969 all_env = {**env_vars, **os.environ} 

970 

971 # Check if required email vars are present 

972 smtp_host = all_env.get("NX_EMAIL_SMTP_HOST") 

973 sender = all_env.get("NX_EMAIL_SENDER") 

974 recipients_str = all_env.get("NX_EMAIL_RECIPIENTS") 

975 

976 if not all([smtp_host, sender, recipients_str]): 

977 return None # Email not configured 

978 

979 recipients = [r.strip() for r in recipients_str.split(",")] 

980 

981 config_dict = { 

982 "smtp_host": smtp_host, 

983 "sender": sender, 

984 "recipients": recipients, 

985 } 

986 

987 # Add optional fields 

988 if smtp_port := all_env.get("NX_EMAIL_SMTP_PORT"): 

989 config_dict["smtp_port"] = int(smtp_port) 

990 if smtp_username := all_env.get("NX_EMAIL_SMTP_USERNAME"): 

991 config_dict["smtp_username"] = smtp_username 

992 if smtp_password := all_env.get("NX_EMAIL_SMTP_PASSWORD"): 

993 config_dict["smtp_password"] = smtp_password 

994 if use_tls := all_env.get("NX_EMAIL_USE_TLS"): 

995 config_dict["use_tls"] = use_tls.lower() in ("true", "1", "yes") 

996 

997 try: 

998 return EmailConfig(**config_dict) 

999 except ValidationError: 

1000 _logger.exception("Invalid email configuration") 

1001 return None 

1002 

1003 

1004class _SettingsManager: 

1005 """ 

1006 Internal manager for the settings singleton. 

1007 

1008 Provides a refresh mechanism for testing while maintaining 

1009 the convenient import pattern for production code. 

1010 """ 

1011 

1012 def __init__(self): 

1013 self._settings: Settings | None = None 

1014 

1015 def get(self) -> Settings: 

1016 """Get the current settings instance, creating if needed.""" 

1017 if self._settings is None: 

1018 self._settings = self._create() 

1019 return self._settings 

1020 

1021 def _create(self) -> Settings: 

1022 """Create a new Settings instance.""" 

1023 try: 

1024 return Settings() 

1025 except ValidationError as e: 

1026 # Add help message to exception using add_note (Python 3.11+) 

1027 # This appears after the exception traceback 

1028 # Strip .dev* suffix from version for documentation link 

1029 doc_version = re.sub(r"\.dev.*$", "", __version__) 

1030 help_msg = ( 

1031 "\n" + "=" * 80 + "\n" 

1032 "NexusLIMS configuration validation failed.\n\n" 

1033 "Quick fix: Run 'nexuslims config edit' to interactively\n" 

1034 " configure NexusLIMS in a terminal UI.\n\n" 

1035 "Reference: https://datasophos.github.io/NexusLIMS/" 

1036 f"{doc_version}/user_guide/configuration.html\n" + "=" * 80 

1037 ) 

1038 if hasattr(e, "add_note"): 

1039 e.add_note(help_msg) 

1040 raise 

1041 

1042 def refresh(self) -> Settings: 

1043 """ 

1044 Refresh settings from current environment variables. 

1045 

1046 Creates a new Settings instance and replaces the cached singleton. 

1047 Primarily used in testing when environment variables are modified. 

1048 

1049 Returns 

1050 ------- 

1051 Settings 

1052 The newly created settings instance 

1053 

1054 Examples 

1055 -------- 

1056 >>> import os 

1057 >>> from nexusLIMS.config import settings, refresh_settings 

1058 >>> 

1059 >>> # In a test, modify environment 

1060 >>> os.environ["NX_FILE_STRATEGY"] = "inclusive" 

1061 >>> 

1062 >>> # Refresh to pick up changes 

1063 >>> refresh_settings() 

1064 >>> 

1065 >>> assert settings.NX_FILE_STRATEGY == "inclusive" 

1066 """ 

1067 self._settings = self._create() 

1068 return self._settings 

1069 

1070 def clear(self) -> None: 

1071 """ 

1072 Clear the settings cache. 

1073 

1074 The next access to settings will create a new instance. 

1075 This is equivalent to refresh() but doesn't immediately create 

1076 a new instance. 

1077 """ 

1078 self._settings = None 

1079 

1080 

1081if TYPE_CHECKING: 

1082 # For type checkers, make the proxy look like Settings 

1083 # This gives us proper type hints and autocomplete 

1084 class _SettingsProxy(Settings): # type: ignore[misc] 

1085 """Type stub for the settings proxy.""" 

1086 

1087else: 

1088 

1089 class _SettingsProxy: 

1090 """ 

1091 Proxy that provides attribute access to the current settings instance. 

1092 

1093 This allows settings to be used like a normal object while supporting 

1094 the refresh mechanism for testing. 

1095 """ 

1096 

1097 def __getattr__(self, name: str): 

1098 # Get the attribute from the actual Settings instance 

1099 attr = getattr(_manager.get(), name)