add application ordertype endpoint (#3864)(minor)

## Description
Closes Clinical-Genomics/add-new-tech#130
Adds the endpoint requesting applications given the order types

### Added

- Endpoint function `get_application_order_types` in `cg/server/endpoints/applications.py`
- Error handler for endpoint in `cg/server/endpoints/error_handler.py`
- Aplication web service that returns the response to the endpoint in `cg/services/application/service.py`
- ApplicationResponse model
- Join application order query function `_get_join_application_ordertype_query` in `cg/store/base.py`
- CRUD function `link_order_types_to_application`. to create `OrderTypeApplication` entries (create)
- CRUF function `get_active_applications_by_order_type` (read)
- Status filter module `cg/store/filters/status_ordertype_application_filters.py` for OrderTypeApplication table and one filter function
- Attribute `order_types` to the `Appliacation` model in the database
- Attribute `application` to the `OrderTypeApplication` model in the database
- Tests for the filter function
- Test for the CRUD (read) function
- Fixtures
- Store helpers function
Co-authored-by: Isak Ohlsson Ångnell <[email protected]>

cg/server/endpoints/applications.py

@@ -3,12 +3,12 @@
 
 from flask import Blueprint, abort, jsonify, make_response
 
+from cg.models.orders.constants import OrderType
+from cg.server.endpoints.error_handler import handle_missing_entries
 from cg.server.endpoints.utils import before_request, is_public
-from cg.server.ext import db
-from cg.store.models import (
-    Application,
-    ApplicationLimitations,
-)
+from cg.server.ext import applications_service, db
+from cg.services.application.models import ApplicationResponse
+from cg.store.models import Application, ApplicationLimitations
 
 APPLICATIONS_BLUEPRINT = Blueprint("applications", __name__, url_prefix="/api/v1")
 APPLICATIONS_BLUEPRINT.before_request(before_request)
@@ -23,6 +23,14 @@ def get_applications():
     return jsonify(applications=parsed_applications)
 
 
+@APPLICATIONS_BLUEPRINT.route("/applications/<order_type>")
+@handle_missing_entries
+def get_application_order_types(order_type: OrderType):
+    """Return application order types."""
+    applications: ApplicationResponse = applications_service.get_valid_applications(order_type)
+    return jsonify(applications.model_dump())
+
+
 @APPLICATIONS_BLUEPRINT.route("/applications/<tag>")
 @is_public
 def get_application(tag: str):

cg/server/endpoints/error_handler.py

@@ -0,0 +1,27 @@
+import logging
+from functools import wraps
+from http import HTTPStatus
+
+from flask import jsonify
+
+from cg.store.exc import EntryNotFoundError
+
+LOG = logging.getLogger(__name__)
+
+
+def handle_missing_entries(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 endpoint: {error}")
+            return (
+                jsonify(error="An error occurred while processing your request."),
+                HTTPStatus.INTERNAL_SERVER_ERROR,
+            )
+
+    return wrapper

cg/server/ext.py

@@ -10,11 +10,10 @@
 from cg.clients.freshdesk.freshdesk_client import FreshdeskClient
 from cg.meta.orders.ticket_handler import TicketHandler
 from cg.server.app_config import app_config
+from cg.services.application.service import ApplicationsWebService
 from cg.services.delivery_message.delivery_message_service import DeliveryMessageService
 from cg.services.orders.order_service.order_service import OrderService
-from cg.services.orders.order_summary_service.order_summary_service import (
-    OrderSummaryService,
-)
+from cg.services.orders.order_summary_service.order_summary_service import OrderSummaryService
 from cg.services.orders.submitters.order_submitter_registry import (
     OrderSubmitterRegistry,
     setup_order_submitter_registry,
@@ -86,6 +85,7 @@ def init_app(self, app):
 
 admin = Admin(name="Clinical Genomics")
 lims = FlaskLims()
+applications_service = ApplicationsWebService(store=db)
 analysis_client = AnalysisClient()
 delivery_message_service = DeliveryMessageService(store=db, trailblazer_api=analysis_client)
 summary_service = OrderSummaryService(store=db, analysis_client=analysis_client)

cg/services/application/models.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class ApplicationResponse(BaseModel):
+    applications: list[str]

cg/services/application/service.py

@@ -0,0 +1,21 @@
+from cg.models.orders.constants import OrderType
+from cg.services.application.models import ApplicationResponse
+from cg.store.models import Application
+from cg.store.store import Store
+
+
+def create_application_response(app_tags: list[str]) -> ApplicationResponse:
+    return ApplicationResponse(applications=app_tags)
+
+
+class ApplicationsWebService:
+
+    def __init__(self, store: Store):
+        self.store = store
+
+    def get_valid_applications(self, order_type: OrderType) -> ApplicationResponse:
+        applications: list[Application] = self.store.get_active_applications_by_order_type(
+            order_type
+        )
+        app_tags: list[str] = [application.tag for application in applications]
+        return create_application_response(app_tags)

cg/store/base.py

@@ -6,12 +6,7 @@
 from sqlalchemy import and_, func
 from sqlalchemy.orm import Query, Session
 
-from cg.store.models import (
-    Analysis,
-    Application,
-    ApplicationLimitations,
-    ApplicationVersion,
-)
+from cg.store.models import Analysis, Application, ApplicationLimitations, ApplicationVersion
 from cg.store.models import Base as ModelBase
 from cg.store.models import (
     Case,
@@ -85,6 +80,10 @@ def _get_join_sample_case_order_query(self) -> Query:
             self._get_query(table=Sample).join(Case.links).join(CaseSample.sample).join(Case.orders)
         )
 
+    def _get_join_application_ordertype_query(self) -> Query:
+        """Return join application to order type query."""
+        return self._get_query(table=Application).join(Application.order_types)
+
     def _get_join_sample_application_version_query(self) -> Query:
         """Return join sample to application version query."""
         return (

cg/store/crud/create.py

@@ -7,6 +7,7 @@
 
 from cg.constants import DataDelivery, Priority, Workflow
 from cg.constants.archiving import PDC_ARCHIVE_LOCATION
+from cg.models.orders.constants import OrderType
 from cg.models.orders.order import OrderIn
 from cg.services.illumina.data_transfer.models import (
     IlluminaFlowCellDTO,
@@ -37,6 +38,7 @@
     IlluminaSequencingRun,
     Invoice,
     Order,
+    OrderTypeApplication,
     Organism,
     PacbioSampleSequencingMetrics,
     PacbioSequencingRun,
@@ -128,6 +130,15 @@ def add_application(
             **kwargs,
         )
 
+    def link_order_types_to_application(
+        self, application: Application, order_types: list[OrderType]
+    ) -> list[OrderTypeApplication]:
+        new_orders: list = []
+        for order_type in order_types:
+            new_record = OrderTypeApplication(application=application, order_type=order_type)
+            new_orders.append(new_record)
+        return new_orders
+
     def add_application_version(
         self,
         application: Application,

cg/store/crud/read.py

@@ -11,19 +11,11 @@
 from cg.constants.constants import CaseActions, CustomerId, PrepCategory, SampleType
 from cg.exc import CaseNotFoundError, CgError, OrderNotFoundError, SampleNotFoundError
 from cg.server.dto.orders.orders_request import OrdersRequest
-from cg.server.dto.samples.collaborator_samples_request import (
-    CollaboratorSamplesRequest,
-)
+from cg.server.dto.samples.collaborator_samples_request import CollaboratorSamplesRequest
 from cg.store.base import BaseHandler
 from cg.store.exc import EntryNotFoundError
-from cg.store.filters.status_analysis_filters import (
-    AnalysisFilter,
-    apply_analysis_filter,
-)
-from cg.store.filters.status_application_filters import (
-    ApplicationFilter,
-    apply_application_filter,
-)
+from cg.store.filters.status_analysis_filters import AnalysisFilter, apply_analysis_filter
+from cg.store.filters.status_application_filters import ApplicationFilter, apply_application_filter
 from cg.store.filters.status_application_limitations_filters import (
     ApplicationLimitationsFilter,
     apply_application_limitations_filter,
@@ -33,23 +25,14 @@
     apply_application_versions_filter,
 )
 from cg.store.filters.status_bed_filters import BedFilter, apply_bed_filter
-from cg.store.filters.status_bed_version_filters import (
-    BedVersionFilter,
-    apply_bed_version_filter,
-)
+from cg.store.filters.status_bed_version_filters import BedVersionFilter, apply_bed_version_filter
 from cg.store.filters.status_case_filters import CaseFilter, apply_case_filter
-from cg.store.filters.status_case_sample_filters import (
-    CaseSampleFilter,
-    apply_case_sample_filter,
-)
+from cg.store.filters.status_case_sample_filters import CaseSampleFilter, apply_case_sample_filter
 from cg.store.filters.status_collaboration_filters import (
     CollaborationFilter,
     apply_collaboration_filter,
 )
-from cg.store.filters.status_customer_filters import (
-    CustomerFilter,
-    apply_customer_filter,
-)
+from cg.store.filters.status_customer_filters import CustomerFilter, apply_customer_filter
 from cg.store.filters.status_illumina_flow_cell_filters import (
     IlluminaFlowCellFilter,
     apply_illumina_flow_cell_filters,
@@ -64,10 +47,11 @@
 )
 from cg.store.filters.status_invoice_filters import InvoiceFilter, apply_invoice_filter
 from cg.store.filters.status_order_filters import OrderFilter, apply_order_filters
-from cg.store.filters.status_organism_filters import (
-    OrganismFilter,
-    apply_organism_filter,
+from cg.store.filters.status_ordertype_application_filters import (
+    OrderTypeApplicationFilter,
+    apply_order_type_application_filter,
 )
+from cg.store.filters.status_organism_filters import OrganismFilter, apply_organism_filter
 from cg.store.filters.status_pacbio_smrt_cell_filters import (
     PacBioSMRTCellFilter,
     apply_pac_bio_smrt_cell_filters,
@@ -838,6 +822,25 @@ def get_applications(self) -> list[Application]:
             .all()
         )
 
+    def get_active_applications_by_order_type(self, order_type: str) -> list[Application]:
+        """
+        Return all possible non-archived applications for an order type.
+        Raises:
+            EntryNotFoundError: If no applications are found for the order type.
+        """
+        non_archived_applications: Query = apply_application_filter(
+            applications=self._get_join_application_ordertype_query(),
+            filter_functions=[ApplicationFilter.IS_NOT_ARCHIVED],
+        )
+        applications: list[Application] = apply_order_type_application_filter(
+            order_type_applications=non_archived_applications,
+            filter_functions=[OrderTypeApplicationFilter.BY_ORDER_TYPE],
+            order_type=order_type,
+        ).all()
+        if not applications:
+            raise EntryNotFoundError(f"No applications found for order type {order_type}")
+        return applications
+
     def get_current_application_version_by_tag(self, tag: str) -> ApplicationVersion | None:
         """Return the current application version for an application tag."""
         application = self.get_application_by_tag(tag=tag)

cg/store/filters/status_ordertype_application_filters.py

@@ -0,0 +1,34 @@
+from enum import Enum
+
+from sqlalchemy.orm import Query
+
+from cg.models.orders.constants import OrderType
+from cg.store.models import OrderTypeApplication
+
+
+def filter_applications_by_order_type(
+    order_type_applications: Query, order_type: OrderType, **kwargs
+) -> Query:
+    """Return application by order type."""
+    return order_type_applications.filter(OrderTypeApplication.order_type == order_type)
+
+
+def apply_order_type_application_filter(
+    filter_functions: list[callable],
+    order_type_applications: Query,
+    order_type: OrderType = None,
+) -> Query:
+    """Apply filtering functions to the ordertype_applications query and return filtered results."""
+
+    for filter_function in filter_functions:
+        order_type_applications: Query = filter_function(
+            order_type_applications=order_type_applications,
+            order_type=order_type,
+        )
+    return order_type_applications
+
+
+class OrderTypeApplicationFilter(Enum):
+    """Define OrderTypeApplication filter functions."""
+
+    BY_ORDER_TYPE = filter_applications_by_order_type

cg/store/models.py

@@ -181,6 +181,9 @@ class Application(Base):
     pipeline_limitations: Mapped[list["ApplicationLimitations"]] = orm.relationship(
         back_populates="application"
     )
+    order_types: Mapped[list["OrderTypeApplication"]] = orm.relationship(
+        "OrderTypeApplication", back_populates="application"
+    )
 
     def __str__(self) -> str:
         return self.tag
@@ -1202,4 +1205,4 @@ class OrderTypeApplication(Base):
     application_id: Mapped[int] = mapped_column(
         ForeignKey("application.id", ondelete="CASCADE"), primary_key=True
     )
-    application: Mapped[Application] = orm.relationship("Application")
+    application: Mapped[Application] = orm.relationship("Application", back_populates="order_types")

tests/store/crud/read/test_read_application.py

@@ -0,0 +1,23 @@
+import pytest
+
+from cg.models.orders.constants import OrderType
+from cg.store.exc import EntryNotFoundError
+from cg.store.models import Application
+from cg.store.store import Store
+
+
+def test_get_active_applications_by_order_type_no_application(store: Store):
+    """Test that if there are not applications for a given type an error is raised."""
+    # GIVEN a store with applications without order types
+    applications: list[Application] = store.get_applications()
+    for application in applications:
+        assert not application.order_types
+
+    # GIVEN an order type
+    order_type: OrderType = OrderType.PACBIO_LONG_READ
+
+    # WHEN getting active applications by order type
+    with pytest.raises(EntryNotFoundError):
+        store.get_active_applications_by_order_type(order_type)
+
+    # THEN an EntryNotFoundError is raised

tests/store/filters/test_status_order_type_application_filters.py

@@ -0,0 +1,36 @@
+from sqlalchemy.orm import Query
+
+from cg.models.orders.constants import OrderType
+from cg.store.filters.status_ordertype_application_filters import filter_applications_by_order_type
+from cg.store.models import Application, OrderTypeApplication
+from cg.store.store import Store
+from tests.store_helpers import StoreHelpers
+
+
+def test_filter_applications_by_order_type(applications_store: Store, helpers: StoreHelpers):
+    # GIVEN a store with applications
+    applications: list[Application] = applications_store.get_applications()
+    assert applications
+
+    # GIVEN an order type
+    order_type = OrderType.PACBIO_LONG_READ
+
+    # GIVEN that one application has the given order type
+    helpers.add_application_order_type(
+        store=applications_store, application=applications[0], order_types=[order_type]
+    )
+
+    # GIVEN that another application has a different order type
+    helpers.add_application_order_type(
+        store=applications_store, application=applications[1], order_types=[OrderType.BALSAMIC]
+    )
+
+    # WHEN filtering applications by order type
+    order_type_applications: Query = applications_store._get_query(table=OrderTypeApplication)
+    filtered_order_type_applications: Query = filter_applications_by_order_type(
+        order_type_applications=order_type_applications, order_type=order_type
+    )
+
+    # THEN assert that only the applications with the given order type are returned
+    assert filtered_order_type_applications.count() == 1
+    assert filtered_order_type_applications.first().order_type == order_type

tests/store_helpers.py

@@ -36,6 +36,7 @@
     IlluminaSequencingRun,
     Invoice,
     Order,
+    OrderTypeApplication,
     Organism,
     Panel,
     Pool,
@@ -251,6 +252,18 @@ def add_application(
         store.session.commit()
         return application
 
+    def add_application_order_type(
+        self, store: Store, application: Application, order_types: list[str]
+    ):
+        """Add an order type to an application."""
+        order_app_links: list[OrderTypeApplication] = store.link_order_types_to_application(
+            application=application, order_types=order_types
+        )
+        for link in order_app_links:
+            store.session.add(link)
+        store.session.commit()
+        return order_app_links
+
     @staticmethod
     def ensure_application_limitation(
         store: Store,