From 9d1c7ceffb3064b9e066c6bbdb96736b3218f147 Mon Sep 17 00:00:00 2001 From: mgcam Date: Tue, 5 Mar 2024 12:05:46 +0000 Subject: [PATCH] pydantic BaseModel is replaced by pydantic dataclass ... for some models in order to simplify instantiation of the objects. --- lang_qc/db/helper/wells.py | 21 ++----- lang_qc/endpoints/pacbio_well.py | 5 +- lang_qc/models/pacbio/well.py | 98 ++++++++++++++++---------------- tests/test_pac_well_full.py | 12 +++- 4 files changed, 67 insertions(+), 69 deletions(-) diff --git a/lang_qc/db/helper/wells.py b/lang_qc/db/helper/wells.py index 63ab3b77..91c1d9b7 100644 --- a/lang_qc/db/helper/wells.py +++ b/lang_qc/db/helper/wells.py @@ -290,14 +290,7 @@ def _get_wells_for_status( id_product = qc_state_model.id_product mlwh_well = self.get_mlwh_well_by_product_id(id_product=id_product) if mlwh_well is not None: - pbw = PacBioWell( - id_product=id_product, - run_name=mlwh_well.pac_bio_run_name, - plate_number=mlwh_well.plate_number, - label=mlwh_well.well_label, - qc_state=qc_state_model, - ) - pbw.copy_run_tracking_info(mlwh_well) + pbw = PacBioWell(db_well=mlwh_well, qc_state=qc_state_model) wells.append(pbw) else: """ @@ -398,16 +391,10 @@ def _well_models( pb_wells = [] for db_well in db_wells_list: id_product = db_well.id_pac_bio_product - attrs = { - "id_product": id_product, - "run_name": db_well.pac_bio_run_name, - "plate_number": db_well.plate_number, - "label": db_well.well_label, - } + qc_state = None if id_product in qced_products: - attrs["qc_state"] = qced_products[id_product][0] - pb_well = PacBioWell.model_validate(attrs) - pb_well.copy_run_tracking_info(db_well) + qc_state = qced_products[id_product][0] + pb_well = PacBioWell(db_well=db_well, qc_state=qc_state) pb_wells.append(pb_well) return pb_wells diff --git a/lang_qc/endpoints/pacbio_well.py b/lang_qc/endpoints/pacbio_well.py index dca152e9..f9d49573 100644 --- a/lang_qc/endpoints/pacbio_well.py +++ b/lang_qc/endpoints/pacbio_well.py @@ -29,6 +29,7 @@ from lang_qc.db.helper.qc import ( assign_qc_state_to_product, claim_qc_for_product, + get_qc_state_for_product, product_has_qc_state, ) from lang_qc.db.helper.well import well_seq_product_find_or_create @@ -179,7 +180,9 @@ def get_seq_metrics( mlwh_well = _find_well_product_or_error(id_product, mlwhdb_session) - return PacBioWellFull.from_orm(mlwh_well, qcdb_session) + qc_state_db = get_qc_state_for_product(session=qcdb_session, id_product=id_product) + qc_state = None if qc_state_db is None else QcState.from_orm(qc_state_db) + return PacBioWellFull(db_well=mlwh_well, qc_state=qc_state) @router.post( diff --git a/lang_qc/models/pacbio/well.py b/lang_qc/models/pacbio/well.py index 9e42a255..efd5abcd 100644 --- a/lang_qc/models/pacbio/well.py +++ b/lang_qc/models/pacbio/well.py @@ -21,12 +21,11 @@ # this program. If not, see . from datetime import datetime -from typing import Optional +from typing import Any, Optional -from pydantic import BaseModel, ConfigDict, Field -from sqlalchemy.orm import Session +from pydantic import Field, model_validator +from pydantic.dataclasses import dataclass -from lang_qc.db.helper.qc import get_qc_state_for_product from lang_qc.db.mlwh_schema import PacBioRunWellMetrics from lang_qc.models.pacbio.experiment import PacBioExperiment from lang_qc.models.pacbio.qc_data import QCDataWell @@ -34,7 +33,8 @@ from lang_qc.models.qc_state import QcState -class PacBioWell(BaseModel, extra="forbid"): +@dataclass +class PacBioWell: """ A response model for a single PacBio well on a particular PacBio run. The class contains the attributes that uniquely define this well (`run_name` @@ -45,6 +45,8 @@ class PacBioWell(BaseModel, extra="forbid"): sequenced or QC metrics or assessment for such data. """ + db_well: PacBioRunWellMetrics = Field(init_var=True) + # Well identifies. id_product: str = Field(title="Product identifier") label: str = Field(title="Well label", description="The label of the PacBio well") @@ -80,19 +82,33 @@ class PacBioWell(BaseModel, extra="forbid"): """, ) - def copy_run_tracking_info(self, db_well: PacBioRunWellMetrics): + @model_validator(mode="before") + def pre_root(cls, values: dict[str, Any]) -> dict[str, Any]: """ Populates this object with the run and well tracking information from a database row that is passed as an argument. """ - self.run_start_time = db_well.run_start - self.run_complete_time = db_well.run_complete - self.well_start_time = db_well.well_start - self.well_complete_time = db_well.well_complete - self.run_status = db_well.run_status - self.well_status = db_well.well_status - self.instrument_name = db_well.instrument_name - self.instrument_type = db_well.instrument_type + + # https://github.com/pydantic/pydantic-core/blob/main/python/pydantic_core/_pydantic_core.pyi + mlwh_db_row: PacBioRunWellMetrics = values.kwargs["db_well"] + assigned = dict() + assigned["id_product"] = mlwh_db_row.id_pac_bio_product + assigned["label"] = mlwh_db_row.well_label + assigned["plate_number"] = mlwh_db_row.plate_number + assigned["run_name"] = mlwh_db_row.pac_bio_run_name + assigned["run_start_time"] = mlwh_db_row.run_start + assigned["run_complete_time"] = mlwh_db_row.run_complete + assigned["well_start_time"] = mlwh_db_row.well_start + assigned["well_complete_time"] = mlwh_db_row.well_complete + assigned["run_status"] = mlwh_db_row.run_status + assigned["well_status"] = mlwh_db_row.well_status + assigned["instrument_name"] = mlwh_db_row.instrument_name + assigned["instrument_type"] = mlwh_db_row.instrument_type + + if "qc_state" in values.kwargs: + assigned["qc_state"] = values.kwargs["qc_state"] + + return assigned class PacBioPagedWells(PagedResponse, extra="forbid"): @@ -110,6 +126,7 @@ class PacBioPagedWells(PagedResponse, extra="forbid"): ) +@dataclass class PacBioWellFull(PacBioWell): """ A response model for a single PacBio well on a particular PacBio run. @@ -128,37 +145,22 @@ class PacBioWellFull(PacBioWell): Laboratory experiment tracking information for this well, if available. """, ) - model_config = ConfigDict(from_attributes=True, extra="forbid") - - @classmethod - def from_orm(cls, mlwh_db_row: PacBioRunWellMetrics, qc_session: Session): - - id_product = mlwh_db_row.id_pac_bio_product - obj = cls( - id_product=id_product, - run_name=mlwh_db_row.pac_bio_run_name, - plate_number=mlwh_db_row.plate_number, - label=mlwh_db_row.well_label, - metrics=QCDataWell.from_orm(mlwh_db_row), - ) - obj.copy_run_tracking_info(mlwh_db_row) - - experiment_info = [] - for row in mlwh_db_row.pac_bio_product_metrics: - exp_row = row.pac_bio_run - if exp_row: - experiment_info.append(exp_row) - else: - # Do not supply incomplete data. - experiment_info = [] - break - if len(experiment_info): - obj.experiment_tracking = PacBioExperiment.from_orm(experiment_info) - - qc_state_db = get_qc_state_for_product( - session=qc_session, id_product=id_product - ) - if qc_state_db is not None: - obj.qc_state = QcState.from_orm(qc_state_db) - - return obj + + @model_validator(mode="before") + def pre_root(cls, values: dict[str, Any]) -> dict[str, Any]: + + assigned = super().pre_root(values) + mlwh_db_row: PacBioRunWellMetrics = values.kwargs["db_well"] + + assigned["metrics"] = QCDataWell.from_orm(mlwh_db_row) + + product_metrics = mlwh_db_row.pac_bio_product_metrics + experiment_info = [ + pbr for pbr in [pm.pac_bio_run for pm in product_metrics] if pbr is not None + ] + # Occasionally product rows are not linked to LIMS rows. + # Go for all or nothing, do not supply incomplete data. + if len(experiment_info) and (len(experiment_info) == len(product_metrics)): + assigned["experiment_tracking"] = PacBioExperiment.from_orm(experiment_info) + + return assigned diff --git a/tests/test_pac_well_full.py b/tests/test_pac_well_full.py index b1a700e6..c8ff08cb 100644 --- a/tests/test_pac_well_full.py +++ b/tests/test_pac_well_full.py @@ -1,5 +1,6 @@ from npg_id_generation.pac_bio import PacBioEntity +from lang_qc.db.helper.qc import get_qc_states_by_id_product_list from lang_qc.db.helper.wells import WellWh from lang_qc.models.pacbio.well import PacBioWellFull from tests.conftest import compare_dates, insert_from_yaml @@ -21,7 +22,7 @@ def test_creating_experiment_object( ).hash_product_id() well_row = helper.get_mlwh_well_by_product_id(id_product) - pb_well = PacBioWellFull.from_orm(well_row, qcdb_test_session) + pb_well = PacBioWellFull(db_well=well_row) assert pb_well.id_product == id_product assert pb_well.run_name == "TRACTION-RUN-92" assert pb_well.label == "A1" @@ -45,7 +46,12 @@ def test_creating_experiment_object( ).hash_product_id() well_row = helper.get_mlwh_well_by_product_id(id_product) - pb_well = PacBioWellFull.from_orm(well_row, qcdb_test_session) + qc_state = get_qc_states_by_id_product_list( + session=qcdb_test_session, + ids=[id_product], + sequencing_outcomes_only=True, + ) + pb_well = PacBioWellFull(db_well=well_row, qc_state=qc_state) assert pb_well.id_product == id_product assert pb_well.run_name == "TRACTION_RUN_1" assert pb_well.label == "B1" @@ -65,7 +71,7 @@ def test_creating_experiment_object( ).hash_product_id() well_row = helper.get_mlwh_well_by_product_id(id_product) - pb_well = PacBioWellFull.from_orm(well_row, qcdb_test_session) + pb_well = PacBioWellFull(db_well=well_row, qc_state=None) assert pb_well.id_product == id_product assert pb_well.run_name == "TRACTION_RUN_10" assert pb_well.label == "C1"