nexusLIMS.db.migrations.utils#

Utilities for database migrations.

Provides helper functions for data integrity verification and backup creation that can be used by migration scripts.

Module Contents#

Functions#

create_backup

Create timestamped backup of database before migration.

verify_table_integrity

Verify table data was preserved during migration.

API#

nexusLIMS.db.migrations.utils.create_backup(connection) → Path[source]#

Create timestamped backup of database before migration.

Parameters:

connection – SQLAlchemy connection to get database path from

Returns:

Path to the backup file

Return type:

Path

Examples:

>>> from alembic import op
>>> from nexusLIMS.migrations.utils import create_backup
>>> def upgrade():
...     connection = op.get_bind()
...     create_backup(connection)
...     # ... perform migration ...
nexusLIMS.db.migrations.utils.verify_table_integrity(connection, table_name: str, expected_count: int, expected_pk_range: tuple[int, int] | None = None, expected_distribution: dict | None = None, distribution_column: str | None = None, pk_column: str = 'id')[source]#

Verify table data was preserved during migration.

Parameters:

  • connection – SQLAlchemy connection for queries

  • table_name (str) – Name of the table to verify

  • expected_count (int) – Expected number of rows

  • expected_pk_range (tuple[int, int] | None) – Expected (min, max) primary key values

  • expected_distribution (dict | None) – Expected distribution of values in a column (e.g., status counts)

  • distribution_column (str | None) – Column name for distribution check

  • pk_column (str) – Primary key column name (default: “id”)

Raises:

RuntimeError – If data integrity checks fail

Examples:

>>> from alembic import op
>>> from nexusLIMS.migrations.utils import verify_table_integrity
>>> def upgrade():
...     connection = op.get_bind()
...     # Before migration: collect baseline
...     result = connection.execute(sa.text("SELECT COUNT(*) FROM my_table"))
...     count = result.scalar()
...     # After migration: verify
...     verify_table_integrity(connection, "my_table_new", count)