Coverage for nexusLIMS/cli/migrate.py: 100%

1# ruff: noqa: PLC0415

2"""CLI for NexusLIMS database migrations.

4Provides simple commands for common database migration operations, while

5still allowing advanced users to access the underlying Alembic functionality.

7Usage

8-----

9.. code-block:: bash

11 # Initialize a new database

12 nexuslims db init

14 # Upgrade to latest schema version

15 nexuslims db upgrade

17 # Show current database version

18 nexuslims db current

20 # Check for pending migrations

21 nexuslims db check

23 # Downgrade one migration

24 nexuslims db downgrade

26 # Browse the database interactively

27 nexuslims db view

29 # Create a demo database with sample instruments

30 nexuslims db create-demo [PATH]

32 # Advanced: Run any Alembic command

33 nexuslims db alembic history --verbose

35Examples

36--------

37Set up a new database:

39.. code-block:: bash

41 nexuslims db init

43Check database status:

45.. code-block:: bash

47 nexuslims db current

48 nexuslims db check

50Apply pending migrations:

52.. code-block:: bash

54 nexuslims db upgrade

56Notes

57-----

58This command automatically locates the migrations directory inside the

59installed package, making it work correctly whether NexusLIMS is installed

60via pip, uv, or run from source.

61"""

63from importlib.resources import files

64from pathlib import Path

67def _get_migrations_dir() -> Path:

68 """Locate the migrations directory inside the installed package.

70 Uses importlib.resources (Python 3.9+) to find nexusLIMS.db.migrations

71 regardless of whether the package is installed normally, as an editable

72 install, or run from source.

74 Returns

75 -------

76 pathlib.Path

77 Absolute path to the nexusLIMS/db/migrations/ directory.

79 Raises

80 ------

81 ImportError

82 If the nexusLIMS.db.migrations package cannot be found.

83 """

84 try:

85 migrations_resource = files("nexusLIMS.db.migrations")

86 return Path(str(migrations_resource))

87 except (ImportError, TypeError) as e:

88 msg = (

89 "Could not locate nexusLIMS.db.migrations package. "

90 "Ensure NexusLIMS is properly installed."

91 )

92 raise ImportError(msg) from e

95def _get_alembic_config():

96 """Create an Alembic Config object for the packaged migrations.

98 Returns

99 -------

100 alembic.config.Config

101 Configured Alembic Config object with script_location set.

102 """

103 from alembic.config import Config

104

105 migrations_dir = _get_migrations_dir()

106 cfg = Config()

107 cfg.set_main_option("script_location", str(migrations_dir))

108 return cfg

109

110

111def _run_alembic_command(command_name: str, *args, **kwargs):

112 """Run an Alembic command programmatically.

113

114 Parameters

115 ----------

116 command_name : str

117 Name of the Alembic command function (e.g., 'upgrade', 'downgrade')

118 *args

119 Positional arguments to pass to the Alembic command

120 **kwargs

121 Keyword arguments to pass to the Alembic command

122 """

123 import alembic.command

124

125 cfg = _get_alembic_config()

126 command_func = getattr(alembic.command, command_name)

127 command_func(cfg, *args, **kwargs)

128

129

130def _get_current_revision() -> str:

131 """Get the current database revision.

132

133 Returns

134 -------

135 str

136 Current revision ID, "none" if not stamped, or "unknown" if database

137 doesn't exist

138 """

139 import os

140

141 from alembic.runtime.migration import MigrationContext

142 from sqlalchemy import create_engine

143

144 db_path = os.getenv("NX_DB_PATH")

145 if not db_path:

146 return "unknown"

147

148 # Check if database file exists

149 if not Path(db_path).exists():

150 return "unknown"

151

152 try:

153 engine = create_engine(f"sqlite:///{db_path}")

154 with engine.connect() as connection:

155 context = MigrationContext.configure(connection)

156 current_rev = context.get_current_revision()

157 return current_rev or "none"

158 except Exception:

159 return "unknown"

160

161

162def _cli(): # noqa: PLR0915

163 """Create the Click CLI application.

164

165 Lazy import of click to speed up --help for other entry points.

166 """

167 import click

168

169 @click.group(invoke_without_command=True)

170 @click.option("--version", is_flag=True, help="Show version and exit")

171 @click.pass_context

172 def cli(ctx, version):

173 """Manage NexusLIMS database.

174

175 This tool provides simple commands for common database operations.

176 For advanced usage, use 'nexuslims db alembic [COMMAND]' to

177 access the full Alembic CLI.

178 """

179 # Load .env file if it exists (before any commands run)

180 from dotenv import find_dotenv, load_dotenv

181

182 load_dotenv(find_dotenv(usecwd=True))

183

184 if version:

185 from nexusLIMS import __version__

186

187 click.echo(f"nexuslims db (NexusLIMS {__version__})")

188 ctx.exit()

189

190 if ctx.invoked_subcommand is None:

191 click.echo(ctx.get_help())

192

193 @cli.command()

194 @click.option(

195 "--force",

196 is_flag=True,

197 help="Overwrite existing database file if it exists",

198 )

199 def init(force):

200 """Initialize a new NexusLIMS database.

201

202 Creates the database file at NX_DB_PATH, applies the schema,

203 and marks it as migrated to the latest version.

204

205 This creates the database file and applies all migrations

206 (equivalent to 'nexuslims db upgrade head' on a new DB).

207 """

208 import os

209 import sys

210

211 # Get DB path from environment variable directly (not from settings)

212 # because settings validation requires the file to exist

213 db_path_str = os.getenv("NX_DB_PATH")

214 if not db_path_str:

215 click.secho(

216 "Error: NX_DB_PATH environment variable is not set.", fg="red", err=True

217 )

218 click.echo("\nSet it via the interactive configurator:\n", err=True)

219 click.echo(" nexuslims config edit\n", err=True)

220 click.echo("Or set it directly, e.g.:\n", err=True)

221 click.echo(" export NX_DB_PATH=/path/to/database.db", err=True)

222 sys.exit(1)

223

224 db_path = Path(db_path_str)

225

226 if db_path.exists() and not force:

227 click.secho(

228 f"Error: Database already exists at {db_path}", fg="red", err=True

229 )

230 click.echo(

231 "Use --force to overwrite, or remove the file manually.", err=True

232 )

233 sys.exit(1)

234

235 # Create parent directory if it doesn't exist

236 db_path.parent.mkdir(parents=True, exist_ok=True)

237

238 # Create empty database file

239 db_path.touch()

240

241 click.echo(f"Initializing database at {db_path}...")

242

243 # Run all migrations to create the schema

244 try:

245 _run_alembic_command("upgrade", "head")

246 click.secho("✓ Database initialized successfully", fg="green")

247 click.echo(f" Current version: {_get_current_revision()}")

248 except Exception as e:

249 click.secho(f"Error initializing database: {e}", fg="red", err=True)

250 # Clean up the empty database file on failure

251 if db_path.exists():

252 db_path.unlink()

253 sys.exit(1)

254

255 @cli.command()

256 @click.argument("revision", default="head")

257 @click.option(

258 "--sql", is_flag=True, help="Generate SQL script instead of applying changes"

259 )

260 def upgrade(revision, sql):

261 r"""Upgrade database to a later version.

262

263 REVISION is the target migration version (default: 'head' for latest).

264

265 \b

266 Examples:

267 nexuslims db upgrade # Upgrade to latest

268 nexuslims db upgrade +1 # Upgrade one version

269 nexuslims db upgrade abc123 # Upgrade to specific revision

270 """

271 import os

272 import sys

273

274 # Check database exists (unless in SQL mode where we just generate SQL)

275 if not sql:

276 db_path = os.getenv("NX_DB_PATH")

277 if not db_path or not Path(db_path).exists():

278 click.secho(

279 "Error: Database does not exist. Run 'nexuslims db init' first.",

280 fg="red",

281 err=True,

282 )

283 sys.exit(1)

284

285 # Get current revision before upgrade

286 before_revision = _get_current_revision()

287

288 try:

289 _run_alembic_command("upgrade", revision, sql=sql)

290 if not sql:

291 # Get new revision after upgrade

292 after_revision = _get_current_revision()

293 click.secho("✓ Database upgraded successfully", fg="green")

294 click.echo(f" {before_revision} -> {after_revision}")

295 except Exception as e:

296 click.secho(f"Error upgrading database: {e}", fg="red", err=True)

297 sys.exit(1)

298

299 @cli.command()

300 @click.argument("revision", default="-1")

301 @click.option(

302 "--sql", is_flag=True, help="Generate SQL script instead of applying changes"

303 )

304 def downgrade(revision, sql):

305 r"""Downgrade database to an earlier version.

306

307 REVISION is the target migration version (default: '-1' for one step back).

308

309 \b

310 Examples:

311 nexuslims db downgrade # Downgrade one version

312 nexuslims db downgrade -2 # Downgrade two versions

313 nexuslims db downgrade abc123 # Downgrade to specific revision

314 """

315 import os

316 import sys

317

318 # Check database exists (unless in SQL mode where we just generate SQL)

319 if not sql:

320 db_path = os.getenv("NX_DB_PATH")

321 if not db_path or not Path(db_path).exists():

322 click.secho(

323 "Error: Database does not exist.",

324 fg="red",

325 err=True,

326 )

327 sys.exit(1)

328

329 # Get current revision before downgrade

330 before_revision = _get_current_revision()

331

332 try:

333 _run_alembic_command("downgrade", revision, sql=sql)

334 if not sql:

335 # Get new revision after downgrade

336 after_revision = _get_current_revision()

337 click.secho("✓ Database downgraded successfully", fg="green")

338 click.echo(f" {before_revision} -> {after_revision}")

339 except Exception as e:

340 click.secho(f"Error downgrading database: {e}", fg="red", err=True)

341 sys.exit(1)

342

343 @cli.command()

344 @click.option("--verbose", "-v", is_flag=True, help="Show detailed information")

345 def current(verbose):

346 """Show the current database migration version.

347

348 Displays the revision ID that the database is currently at.

349 """

350 import os

351 import sys

352

353 # Check database exists

354 db_path = os.getenv("NX_DB_PATH")

355 if not db_path or not Path(db_path).exists():

356 click.secho(

357 "Error: Database does not exist. Run 'nexuslims db init' first.",

358 fg="red",

359 err=True,

360 )

361 sys.exit(1)

362

363 try:

364 _run_alembic_command("current", verbose=verbose)

365 except Exception as e:

366 click.secho(f"Error checking database version: {e}", fg="red", err=True)

367 sys.exit(1)

368

369 @cli.command()

370 def check():

371 """Check if the database has pending migrations.

372

373 Exits with code 0 if database is up-to-date, code 1 if migrations

374 are pending, or code 2 on error.

375 """

376 import os

377 import sys

378

379 from alembic.script import ScriptDirectory

380

381 # Check database exists

382 db_path = os.getenv("NX_DB_PATH")

383 if not db_path or not Path(db_path).exists():

384 click.secho(

385 "Error: Database does not exist. Run 'nexuslims db init' first.",

386 fg="red",

387 err=True,

388 )

389 sys.exit(2)

390

391 try:

392 cfg = _get_alembic_config()

393 script = ScriptDirectory.from_config(cfg)

394

395 # Get current database revision

396 from alembic.runtime.migration import MigrationContext

397 from sqlalchemy import create_engine

398

399 engine = create_engine(f"sqlite:///{db_path}")

400 with engine.connect() as connection:

401 context = MigrationContext.configure(connection)

402 current_rev = context.get_current_revision()

403

404 head_rev = script.get_current_head()

405

406 if current_rev == head_rev:

407 click.secho(

408 f"✓ Database is up-to-date (revision: {current_rev or 'none'})",

409 fg="green",

410 )

411 sys.exit(0)

412 else:

413 click.secho("⚠ Database has pending migrations", fg="yellow", err=True)

414 click.echo(f" Current revision: {current_rev or 'none'}", err=True)

415 click.echo(f" Latest revision: {head_rev}", err=True)

416 click.echo(

417 "\nRun 'nexuslims db upgrade' to apply pending migrations.",

418 err=True,

419 )

420 sys.exit(1)

421 except Exception as e:

422 click.secho(f"Error checking migrations: {e}", fg="red", err=True)

423 sys.exit(2)

424

425 @cli.command()

426 @click.option("--verbose", "-v", is_flag=True, help="Show detailed information")

427 @click.option(

428 "--indicate-current",

429 "-i",

430 is_flag=True,

431 help="Indicate current revision (Alembic default)",

432 )

433 def history(verbose, indicate_current):

434 """Show migration history.

435

436 Displays the revision history for the database migrations.

437 """

438 try:

439 _run_alembic_command(

440 "history", verbose=verbose, indicate_current=indicate_current

441 )

442 except Exception as e:

443 click.secho(f"Error showing history: {e}", fg="red", err=True)

444 import sys

445

446 sys.exit(1)

447

448 @cli.command(

449 context_settings={"ignore_unknown_options": True, "allow_extra_args": True}

450 )

451 @click.pass_context

452 def alembic(ctx):

453 r"""Run Alembic commands directly (advanced usage).

454

455 This passes all arguments directly to Alembic's CLI, allowing

456 access to the full range of Alembic commands and options.

457

458 \b

459 Examples:

460 nexuslims db alembic history --verbose

461 nexuslims db alembic revision --autogenerate -m "Add column"

462 nexuslims db alembic show head

463 """

464 import contextlib

465 import sys

466 import tempfile

467

468 from alembic.config import CommandLine

469

470 migrations_dir = _get_migrations_dir()

471

472 # Create a temporary config file for Alembic

473 with tempfile.NamedTemporaryFile(

474 mode="w", suffix=".ini", delete=False

475 ) as tmp_config:

476 tmp_config.write(f"[alembic]\nscript_location = {migrations_dir}\n")

477 tmp_config_path = Path(tmp_config.name)

478

479 try:

480 # Inject config file and pass through remaining arguments

481 original_argv = sys.argv.copy()

482 sys.argv = [

483 "nexuslims db alembic",

484 "-c",

485 str(tmp_config_path),

486 *ctx.args,

487 ]

488

489 # Run Alembic's CLI

490 CommandLine(prog="nexuslims db alembic").main()

491 finally:

492 sys.argv = original_argv

493 # Clean up the temporary config file

494 with contextlib.suppress(OSError):

495 tmp_config_path.unlink()

496

497 @cli.command("create-demo")

498 @click.argument(

499 "path",

500 type=click.Path(dir_okay=False, path_type=Path),

501 default=None,

502 required=False,

503 )

504 @click.option(

505 "--force",

506 is_flag=True,

507 help="Overwrite the file if it already exists.",

508 )

509 def create_demo(path, force):

510 """Create a demo database pre-populated with sample instruments.

511

512 Creates a SQLite database containing 10 sample instruments with

513 diverse configurations, useful for testing the TUI, generating

514 documentation screenshots, or exploring NexusLIMS without a live

515 NEMO instance.

516

517 PATH defaults to a file named ``nexuslims_demo.db`` in the current

518 working directory when not supplied.

519

520 \b

521 Examples:

522 nexuslims db create-demo

523 nexuslims db create-demo /tmp/demo.db

524 nexuslims db create-demo ./my_demo.db --force

525 """ # noqa: D301

526 import sys

527

528 from nexusLIMS.tui.demo_helpers import create_demo_database

529

530 if path is None:

531 path = Path.cwd() / "nexuslims_demo.db"

532

533 if path.exists() and not force:

534 click.secho(f"Error: file already exists at {path}", fg="red", err=True)

535 click.echo("Use --force to overwrite it.", err=True)

536 sys.exit(1)

537

538 if force and path.exists():

539 path.unlink()

540 click.echo(f"Creating demo database at {path}\u2026")

541 try:

542 create_demo_database(path)

543 click.secho(f"\u2713 Demo database created at {path}", fg="green")

544 except Exception as e:

545 click.secho(f"Error: {e}", fg="red", err=True)

546 sys.exit(1)

547

548 @cli.command()

549 def view():

550 """Open a read-only TUI browser for the NexusLIMS database.

551

552 Launches a Textual-based SQLite browser pointed at the database

553 configured via NX_DB_PATH. You can browse tables and inspect rows

554 across all NexusLIMS tables (instruments, session_log, upload_log,

555 external_user_identifiers).

556

557 \b

558 Keybindings (inside the browser):

559 ----------------------------------

560 - Tab / Shift+Tab Navigate between panels

561 - Enter Select / confirm

562 - q / Ctrl+Q Quit

563 """ # noqa: D301

564 import os

565 import sys

566 from argparse import Namespace

567

568 db_path_str = os.getenv("NX_DB_PATH")

569 if not db_path_str:

570 click.secho(

571 "Error: NX_DB_PATH environment variable is not set.", fg="red", err=True

572 )

573 click.echo(

574 "Run 'nexuslims config edit' or set NX_DB_PATH manually.", err=True

575 )

576 sys.exit(1)

577

578 if not Path(db_path_str).exists():

579 click.secho(

580 f"Error: Database does not exist at {db_path_str}",

581 fg="red",

582 err=True,

583 )

584 click.echo("Run 'nexuslims db init' to create the database.", err=True)

585 sys.exit(1)

586

587 from nexusLIMS.tui.apps.db_browser import NexusLIMSDBApp

588

589 app = NexusLIMSDBApp(Namespace(filepath=db_path_str))

590 app.run()

591

592 return cli

593

594

595def main() -> None:

596 """Entry point for nexuslims db CLI."""

597 cli = _cli()

598 cli()

599

600

601if __name__ == "__main__":

602 main()