Refactor flow cells endpoint (#3549)

cg/exc.py

@@ -282,6 +282,10 @@ class MissingMetrics(CgError):
     """Exception raised when mandatory metrics are missing."""
 
 
+class MissingSequencingMetricsError(CgError):
+    """Exception raised when sequencing metrics are missing."""
+
+
 class ArchiveJobFailedError(CgError):
     """Exception raised when an archival or retrieval job has failed."""
 

cg/server/app.py

@@ -8,8 +8,8 @@
 
 from cg.server import admin, ext, invoices
 from cg.server.app_config import app_config
+from cg.server.endpoints.flow_cells.flow_cells import FLOW_CELLS_BLUEPRINT
 from cg.server.endpoints.analyses import ANALYSES_BLUEPRINT
-from cg.server.endpoints.flow_cells import FLOW_CELLS_BLUEPRINT
 from cg.server.endpoints.orders import ORDERS_BLUEPRINT
 from cg.server.endpoints.applications import APPLICATIONS_BLUEPRINT
 from cg.server.endpoints.cases import CASES_BLUEPRINT

cg/server/endpoints/flow_cells/error_handler.py

@@ -0,0 +1,27 @@
+from http import HTTPStatus
+import logging
+from functools import wraps
+from flask import jsonify
+
+from cg.store.exc import EntryNotFoundError
+
+LOG = logging.getLogger(__name__)
+
+
+def handle_endpoint_errors(func):
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except EntryNotFoundError as error:
+            LOG.error(error)
+            return jsonify(error=str(error)), HTTPStatus.NOT_FOUND
+        except Exception as error:
+            LOG.error(f"Unexpected error in flow cells endpoint: {error}")
+            return (
+                jsonify(error="An error occurred while processing your request."),
+                HTTPStatus.INTERNAL_SERVER_ERROR,
+            )
+
+    return wrapper

cg/server/endpoints/flow_cells/flow_cells.py

@@ -0,0 +1,16 @@
+from flask import Blueprint, jsonify
+
+from cg.server.endpoints.flow_cells.error_handler import handle_endpoint_errors
+from cg.server.endpoints.utils import before_request
+from cg.server.ext import flow_cell_service
+from cg.services.sample_run_metrics_service.dtos import SequencingMetrics
+
+FLOW_CELLS_BLUEPRINT = Blueprint("flowcells", __name__, url_prefix="/api/v1")
+FLOW_CELLS_BLUEPRINT.before_request(before_request)
+
+
+@FLOW_CELLS_BLUEPRINT.route("/flowcells/<flow_cell_name>/sequencing_metrics", methods=["GET"])
+@handle_endpoint_errors
+def get_sequencing_metrics(flow_cell_name: str):
+    metrics: list[SequencingMetrics] = flow_cell_service.get_metrics(flow_cell_name)
+    return jsonify([metric.model_dump() for metric in metrics])

cg/server/ext.py

@@ -9,6 +9,9 @@
 from cg.apps.osticket import OsTicket
 from cg.apps.tb.api import TrailblazerAPI
 from cg.services.delivery_message.delivery_message_service import DeliveryMessageService
+from cg.services.sample_run_metrics_service.sample_run_metrics_service import (
+    SampleRunMetricsService,
+)
 from cg.services.orders.order_service.order_service import OrderService
 from cg.services.orders.order_summary_service.order_summary_service import (
     OrderSummaryService,
@@ -87,6 +90,8 @@ def init_app(self, app):
 summary_service = OrderSummaryService(store=db, analysis_client=analysis_client)
 order_service = OrderService(store=db, status_service=summary_service)
 sample_service = SampleService(db)
+flow_cell_service = SampleRunMetricsService(db)
 order_submitter_registry: OrderSubmitterRegistry = setup_order_submitter_registry(
-    lims=lims, status_db=db
+    lims=lims,
+    status_db=db,
 )

cg/server/dto/sequencing_metrics/sequencing_metrics_request.py → cg/services/sample_run_metrics_service/dtos.py

@@ -1,9 +1,7 @@
-"""DTO for the sequencing metrics."""
-
 from pydantic import BaseModel
 
 
-class SequencingMetricsRequest(BaseModel):
+class SequencingMetrics(BaseModel):
     flow_cell_name: str
     flow_cell_lane_number: int
     sample_internal_id: str

cg/services/sample_run_metrics_service/sample_run_metrics_service.py

@@ -0,0 +1,15 @@
+from cg.services.sample_run_metrics_service.dtos import SequencingMetrics
+from cg.services.sample_run_metrics_service.utils import create_metrics_dto
+from cg.store.models import IlluminaSequencingRun
+from cg.store.store import Store
+
+
+class SampleRunMetricsService:
+    def __init__(self, store: Store):
+        self.store = store
+
+    def get_metrics(self, flow_cell_name: str) -> list[SequencingMetrics]:
+        run: IlluminaSequencingRun = self.store.get_illumina_sequencing_run_by_device_internal_id(
+            flow_cell_name
+        )
+        return create_metrics_dto(run.sample_metrics) if run else []

cg/server/utils.py → cg/services/sample_run_metrics_service/utils.py

@@ -1,14 +1,18 @@
-from cg.server.dto.sequencing_metrics.sequencing_metrics_request import SequencingMetricsRequest
+from cg.services.sample_run_metrics_service.dtos import SequencingMetrics
 from cg.store.models import IlluminaSampleSequencingMetrics
 
 
-def parse_metrics_into_request(
-    metrics: list[IlluminaSampleSequencingMetrics],
-) -> list[SequencingMetricsRequest]:
-    """Parse sequencing metrics into a request."""
-    parsed_metrics: list[SequencingMetricsRequest] = []
+def create_metrics_dto(
+    metrics: list[IlluminaSampleSequencingMetrics] | None,
+) -> list[SequencingMetrics]:
+
+    if not metrics:
+        return []
+
+    parsed_metrics = []
+
     for metric in metrics:
-        parsed_metric = SequencingMetricsRequest(
+        parsed_metric = SequencingMetrics(
             flow_cell_name=metric.instrument_run.device.internal_id,
             flow_cell_lane_number=metric.flow_cell_lane,
             sample_internal_id=metric.sample.internal_id,

tests/services/sample_run_metrics_service/conftest.py

@@ -0,0 +1,13 @@
+import pytest
+
+from cg.services.sample_run_metrics_service.sample_run_metrics_service import (
+    SampleRunMetricsService,
+)
+from cg.store.store import Store
+
+
+@pytest.fixture
+def sample_run_metrics_service(
+    store_with_illumina_sequencing_data: Store,
+) -> SampleRunMetricsService:
+    return SampleRunMetricsService(store_with_illumina_sequencing_data)

tests/services/sample_run_metrics_service/test_sample_metrics_service.py

@@ -0,0 +1,16 @@
+from cg.services.sample_run_metrics_service.sample_run_metrics_service import (
+    SampleRunMetricsService,
+)
+
+
+def test_get_metrics_for_flow_cell(
+    sample_run_metrics_service: SampleRunMetricsService,
+    novaseq_x_flow_cell_id: str,
+):
+    # GIVEN an existing flow cell with sequencing metrics
+
+    # WHEN fetching the metrics for the flow cell
+    metrics = sample_run_metrics_service.get_metrics(novaseq_x_flow_cell_id)
+
+    # THEN the metrics should be returned
+    assert metrics