Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions CHANGELOG
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,26 @@ and this project adheres to [Semantic Versioning](http://semver.org/).

## [Unreleased]

## [1.1.0] 2026-04-23

### Changed

* Previously unused function `get_wafer_content_by_lims_id` is renamed to
`get_records_by_wafer_lims_id` to reflect its functionality and leave room
for retrieval by other LIMS IDs in future. Since the wafers have their own
manufacturer-assigned IDs, it is important to emphasise that the argument
to this function is the ID assigned by LIMS.

In this function the type of the returned object is changed from `list` to
`Sequence` to be consistent with other code in this package.

To follow the existing pattern of naming files with unit tests, tests for
the function are moved from `tests/test_schema.py` to `tests/test_wafer.py`.

### Removed

`tests/test_schema.py`

## [1.0.1] 2026-04-16

### Changed
Expand Down
22 changes: 13 additions & 9 deletions src/npgmlwarehouse/db/wafer.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""
Functions for retrieving the barcoded moieties loaded into one wafer/flowcell/smrtcell
Functions for retrieving the barcoded moieties loaded into one wafer.
"""

from typing import Sequence

from sqlalchemy import select
from sqlalchemy.orm import Session

Expand All @@ -29,23 +31,25 @@
)


def get_wafer_content_by_lims_id(
def get_records_by_wafer_lims_id(
session: Session, id_wafer_lims: str
) -> list[UseqWafer]:
) -> Sequence[UseqWafer]:
"""
Get the LIMS entities that were assigned to a wafer for an Ultimagen run with sample and study metadata.
Get all LIMS entities that correspond to an Ultimagen wafer along with
their sample and study data.

Args:
session (sqlalchemy.orm.Session):
Database session
id_wafer_lims (str): Unique ID for Ultimagen runs which consists of
<batch_for_opentrons>_<id_pool_id>_<runcount>
id_wafer_lims (str): Unique ID for an Ultimagen wafer which consists of
<batch_for_opentrons>_<pool_barcode>_<count>. The count is used to
disambiguate multiple wafers which were processed in the same batch
opentrons process for the same library pool.

Returns:
-------
list[UseqWafer]: Collection of samples from wafer data
Sequence[UseqWafer]: Collection of wafer records
"""
query = (select(UseqWafer).join(Sample).join(Study)).where(
UseqWafer.id_wafer_lims == id_wafer_lims
)
return list(session.scalars(query).all())
return session.scalars(query).all()
40 changes: 0 additions & 40 deletions tests/test_schema.py

This file was deleted.

37 changes: 37 additions & 0 deletions tests/test_wafer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
from pytest import mark as m

from npgmlwarehouse.db.wafer import get_records_by_wafer_lims_id


@m.describe("Test data retrieval for a wafer")
class TestProduct(object):
@m.context("Wafer records are present in `useq_wafer` table")
@m.it("Retrieves correct records")
def test_get_wafer_records(self, testdb):
records = get_records_by_wafer_lims_id(testdb, "122_NT109338I_1")
assert len(records) == 2
assert records[0].id_wafer_lims == "122_NT109338I_1"
assert records[1].id_wafer_lims == "122_NT109338I_1"

metadata = [
(w.sample.uuid_sample_lims, w.id_library_lims, w.study.id_study_lims)
for w in records
]
for md in [
("1788c8d0-6a6c-11e4-8e19-68b59977951c", "SQPU-346269-E:A1", "619"),
("178df8f0-6a6c-11e4-8e19-68b59977951c", "SQPU-346269-E:A2", "619"),
]:
assert md in metadata

records = get_records_by_wafer_lims_id(testdb, "123_NT109345H_1")
assert len(records) == 1
assert records[0].id_wafer_lims == "123_NT109345H_1"
assert (
records[0].sample.uuid_sample_lims == "d41d4a40-a521-11e3-8055-3c4a9275d6c6"
)

@m.context("No records for a given ID in `useq_wafer` table")
@m.it("Returns an empty list")
def test_get_wafer_records_no_record(self, testdb):
records = get_records_by_wafer_lims_id(testdb, "122_NT109338I_2")
assert len(records) == 0