Release 0.2.0

CHANGELOG

@@ -3,6 +3,16 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## [0.2.0] - 2025-12-12
+
+### Added
+
+* get_run_by_id function added to retrieval module
+
+### Fixed
+
+* run.batch_id is now a varchar
+
 ## [0.1.1] - 2025-12-11
 
 ### Fixed

src/npgtracking/db/retrieval.py

@@ -15,6 +15,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from npgtracking.db.schema import (
@@ -66,3 +67,43 @@ def get_runs_by_currentstatus(
         Manufacturer.name == manufacturer_name,
     )
     return query.all()
+
+
+def get_run_by_id(
+    session: Session,
+    batch_id: str | None = None,
+    flowcell_id: str | None = None,
+    id_run: int | None = None,
+) -> Run | None:
+    """
+    Get a Run by its IDs. Either id_run alone, or batch_id and flowcell_id together
+
+    Args:
+      session :
+        Database session
+      ---
+      batch_id :
+        The batch ID from DNA pipelines - must be combined with flowcell_id below
+      flowcell_id :
+        The ID of the flowcell used in the run - must be combined with batch_id
+      ---
+      id_run :
+        NPG Tracking run ID - sufficient on its own
+
+    Returns:
+    -------
+      npgtracking.db.schema.Run or None
+    """
+
+    statement = select(Run)
+
+    if id_run:
+        statement = statement.where(Run.id_run == id_run)
+    elif batch_id and flowcell_id:
+        statement = statement.where(Run.batch_id == batch_id).where(
+            Run.flowcell_id == flowcell_id
+        )
+    else:
+        raise ValueError("Can't get one run without an argument")
+    result = session.execute(statement).scalar_one_or_none()
+    return result

src/npgtracking/db/schema.py

@@ -42,6 +42,7 @@
     with the import from sqlalchemy.dialects.mysql
 """
 
+
 class Base(DeclarativeBase):
     pass
 
@@ -637,7 +638,7 @@ class Run(Base):
     actual_cycle_count: Mapped[Optional[int]] = mapped_column(BIGINT)
     expected_cycle_count: Mapped[Optional[int]] = mapped_column(BIGINT)
     id_run_pair: Mapped[Optional[int]] = mapped_column(BIGINT)
-    batch_id: Mapped[Optional[int]] = mapped_column(BIGINT)
+    batch_id: Mapped[Optional[int]] = mapped_column(String(64))
     flowcell_id: Mapped[Optional[str]] = mapped_column(String(64))
     folder_name: Mapped[Optional[str]] = mapped_column(String(64))
     folder_path_glob: Mapped[Optional[str]] = mapped_column(String(256))

tests/data/db_fixtures/300-Run.yml

@@ -51,3 +51,16 @@
   is_paired: 0
   priority: 1
   team: 'SR'
+- actual_cycle_count: ~
+  batch_id: O44 batch 24798 pool 1
+  expected_cycle_count: ~
+  flowcell_id: 430591
+  folder_name: 430591-20251204_1628
+  folder_path_glob:
+  id_instrument: 130
+  id_instrument_format: 25
+  id_run: 51533
+  id_run_pair: ~
+  is_paired: 0
+  priority: 1
+  team: 'SR'

tests/test_schema.py

@@ -16,8 +16,12 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from pytest import mark as m
+from pytest import raises
 
-from npgtracking.db.retrieval import get_runs_by_currentstatus
+from npgtracking.db.retrieval import (
+    get_run_by_id,
+    get_runs_by_currentstatus,
+)
 
 
 @m.describe("SchemaModel")
@@ -28,7 +32,9 @@ class TestSchemaModel(object):
     def test_schema_no_runs(self, tracking_session):
         status = "run mirrored"
         manufacturer = "Ultima Genomics"
-        tracking_runs = get_runs_by_currentstatus(tracking_session, status, manufacturer)
+        tracking_runs = get_runs_by_currentstatus(
+            tracking_session, status, manufacturer
+        )
         assert len(tracking_runs) == 0
 
     @m.context("When retrieving run records from tracking DB")
@@ -39,7 +45,9 @@ def test_schema_no_runs(self, tracking_session):
     def test_schema_single_run(self, tracking_session):
         status = "run in progress"
         manufacturer = "Ultima Genomics"
-        tracking_runs = get_runs_by_currentstatus(tracking_session, status, manufacturer)
+        tracking_runs = get_runs_by_currentstatus(
+            tracking_session, status, manufacturer
+        )
 
         assert len(tracking_runs) == 1
         run = tracking_runs.pop()
@@ -61,7 +69,9 @@ def test_schema_single_run(self, tracking_session):
     def test_schema_multiple_runs(self, tracking_session):
         status = "off-tool automation in progress"
         manufacturer = "Ultima Genomics"
-        tracking_runs = get_runs_by_currentstatus(tracking_session, status, manufacturer)
+        tracking_runs = get_runs_by_currentstatus(
+            tracking_session, status, manufacturer
+        )
 
         assert len(tracking_runs) == 2
         for run in tracking_runs:
@@ -71,3 +81,24 @@ def test_schema_multiple_runs(self, tracking_session):
             ).pop()
             assert run.instrument_format.manufacturer.name == manufacturer
             assert current_run_status.run_status_dict.description == status
+
+    @m.context("When getting a run by IDs")
+    @m.it("Gives us one or no runs")
+    def test_run_by_id(self, tracking_session):
+        with raises(ValueError, match="Can't get one run without an argument"):
+            get_run_by_id(session=tracking_session, batch_id=None, flowcell_id=None)
+
+        run = get_run_by_id(
+            session=tracking_session,
+            batch_id="O44 batch 24798 pool 1",
+            flowcell_id="430591",
+        )
+        assert run
+        assert run.id_run == 51533
+
+        run = get_run_by_id(session=tracking_session, id_run=51533)
+        assert run
+        assert run.id_run == 51533
+
+        run = get_run_by_id(session=tracking_session, id_run=12345)
+        assert run is None