Add workflow filtering to Orders endpoint (#2881) (patch)

### Added

- Support for filtering orders on workflow.

cg/server/dto/orders/orders_request.py

@@ -3,3 +3,4 @@
 
 class OrdersRequest(BaseModel):
     limit: int | None = None
+    workflow: str | None = None

cg/services/orders/order_service.py

@@ -1,10 +1,9 @@
 from cg.models.orders.order import OrderIn
 from cg.server.dto.orders.orders_request import OrdersRequest
-from cg.server.dto.orders.orders_response import OrdersResponse
 from cg.server.dto.orders.orders_response import Order as OrderResponse
+from cg.server.dto.orders.orders_response import OrdersResponse
 from cg.services.orders.utils import create_order_response, create_orders_response
-from cg.services.orders.utils import create_orders_response
-from cg.store.models import Customer, Order
+from cg.store.models import Order
 from cg.store.store import Store
 
 
@@ -13,9 +12,15 @@ def __init__(self, store: Store) -> None:
         self.store = store
 
     def get_orders(self, orders_request: OrdersRequest) -> OrdersResponse:
-        orders: list[Order] = self.store.get_orders(orders_request.limit)
+        orders: list[Order] = self._get_orders(orders_request)
         return create_orders_response(orders)
 
+    def _get_orders(self, orders_request: OrdersRequest) -> list[Order]:
+        """Returns a list of entries in the table Order."""
+        return self.store.get_orders_by_workflow(
+            workflow=orders_request.workflow, limit=orders_request.limit
+        )
+
     def create_order(self, order_data: OrderIn) -> OrderResponse:
         order: Order = self.store.add_order(order_data)
         return create_order_response(order)

cg/store/crud/read.py

@@ -55,6 +55,7 @@
     SequencingMetricsFilter,
     apply_metrics_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,
@@ -1739,12 +1740,16 @@ def get_all_pools_to_deliver(self) -> list[Pool]:
         )
         return records.all()
 
-    def get_orders(self, limit: int | None = None) -> list[Order]:
-        """Returns a list of entries in the table Order."""
-        records: Query = self._get_query(table=Order)
-        if limit is not None:
-            records = records.limit(limit)
-        return records.all()
+    def get_orders_by_workflow(
+        self, workflow: str | None = None, limit: int | None = None
+    ) -> list[Order]:
+        """Returns a list of entries in Order. The output is filtered on workflow and limited, if given."""
+        orders: Query = self._get_query(table=Order)
+        order_filter_functions: list[Callable] = [OrderFilter.FILTER_ORDERS_BY_WORKFLOW]
+        orders: Query = apply_order_filters(
+            orders=orders, filter_functions=order_filter_functions, workflow=workflow
+        )
+        return orders.limit(limit).all()
 
     def _calculate_estimated_turnaround_time(
         self,

cg/store/filters/status_order_filters.py

@@ -0,0 +1,24 @@
+from enum import Enum
+from typing import Callable
+
+from sqlalchemy.orm import Query
+
+from cg.store.models import Order
+
+
+def filter_orders_by_workflow(orders: Query, workflow: str, **kwargs) -> Query:
+    """Return orders filtered on workflow."""
+    return orders.filter(Order.workflow == workflow) if workflow else orders
+
+
+def apply_order_filters(filter_functions: list[Callable], orders: Query, workflow: str) -> Query:
+    """Apply filtering functions to the order queries and return filtered results."""
+    for filter_function in filter_functions:
+        orders: Query = filter_function(orders=orders, workflow=workflow)
+    return orders
+
+
+class OrderFilter(Enum):
+    """Define order filter functions."""
+
+    FILTER_ORDERS_BY_WORKFLOW: Callable = filter_orders_by_workflow

tests/server/conftest.py

@@ -71,13 +71,25 @@ def order(helpers: StoreHelpers, customer: Customer) -> Order:
 
 
 @pytest.fixture
-def order_another(helpers: StoreHelpers, customer_another) -> Order:
+def order_another(helpers: StoreHelpers, customer_another: Customer) -> Order:
     order: Order = helpers.add_order(
         store=store, customer_id=customer_another.id, ticket_id=2, order_date=datetime.now()
     )
     return order
 
 
+@pytest.fixture
+def order_balsamic(helpers: StoreHelpers, customer_another: Customer) -> Order:
+    order: Order = helpers.add_order(
+        store=store,
+        customer_id=customer_another.id,
+        ticket_id=3,
+        order_date=datetime.now(),
+        workflow=Pipeline.BALSAMIC,
+    )
+    return order
+
+
 @pytest.fixture
 def client(app: Flask) -> Generator[FlaskClient, None, None]:
     # Bypass authentication

tests/server/endpoints/test_orders_endpoint.py

@@ -3,25 +3,39 @@
 import pytest
 from flask.testing import FlaskClient
 
+from cg.constants import Pipeline
 from cg.store.models import Order
 
 
-@pytest.mark.parametrize("limit", [None, 1])
+@pytest.mark.parametrize(
+    "limit, workflow, expected_orders",
+    [
+        (None, Pipeline.MIP_DNA, 2),
+        (1, Pipeline.MIP_DNA, 1),
+        (2, Pipeline.MIP_DNA, 2),
+        (None, Pipeline.BALSAMIC, 1),
+        (None, Pipeline.FLUFFY, 0),
+        (None, None, 3),
+    ],
+)
 def test_orders_endpoint(
-    client: FlaskClient, order: Order, order_another: Order, limit: int | None
+    client: FlaskClient,
+    order: Order,
+    order_another: Order,
+    order_balsamic: Order,
+    limit: int | None,
+    workflow: str,
+    expected_orders: int,
 ):
     """Tests that orders are returned from the orders endpoint"""
-    # GIVEN a store with two orders
+    # GIVEN a store with three orders, two of which are MIP-DNA and the last is BALSAMIC
 
     # WHEN a request is made to get all orders
     endpoint: str = "/api/v1/orders"
-    response = client.get(endpoint, query_string={"limit": limit})
+    response = client.get(endpoint, query_string={"limit": limit, "workflow": workflow})
 
     # THEN the response should be successful
     assert response.status_code == HTTPStatus.OK
 
     # THEN the response contains the correct number of orders
-    if limit:
-        assert len(response.json["orders"]) == 1
-    else:
-        assert len(response.json["orders"]) == 2
+    assert len(response.json["orders"]) == expected_orders

tests/store/crud/conftest.py

@@ -399,3 +399,15 @@ def order_another(helpers: StoreHelpers, store: Store) -> Order:
         store=store, customer_id=2, ticket_id=2, order_date=datetime.now()
     )
     return order
+
+
+@pytest.fixture
+def order_balsamic(helpers: StoreHelpers, store: Store) -> Order:
+    order: Order = helpers.add_order(
+        store=store,
+        customer_id=2,
+        ticket_id=3,
+        order_date=datetime.now(),
+        workflow=Pipeline.BALSAMIC,
+    )
+    return order

tests/store/crud/read/test_read.py

@@ -1529,20 +1529,57 @@ def test_get_orders_empty_store(store: Store):
 
     # WHEN fetching orders
     # THEN none should be returned
-    assert not store.get_orders()
+    assert not store.get_orders_by_workflow()
 
 
 def test_get_orders_populated_store(store: Store, order: Order, order_another: Order):
     # GIVEN a store with two orders
 
     # WHEN fetching orders
     # THEN both should be returned
-    assert len(store.get_orders()) == 2
+    assert len(store.get_orders_by_workflow()) == 2
 
 
 def test_get_orders_limited(store: Store, order: Order, order_another: Order):
-    # GIVEN a store with two orders and a customer
+    # GIVEN a store with two orders
 
     # WHEN fetching a limited amount of orders
     # THEN only one should be returned
-    assert len(store.get_orders(limit=1)) == 1
+    assert len(store.get_orders_by_workflow(limit=1)) == 1
+
+
+def test_get_orders_workflow_filter(
+    store: Store, order: Order, order_another: Order, order_balsamic: Order
+):
+    # GIVEN a store with three orders, one of which is a Balsamic order
+
+    # WHEN fetching only balsamic orders
+    orders: list[Order] = store.get_orders_by_workflow(workflow=Pipeline.BALSAMIC)
+    # THEN only one should be returned
+    assert len(orders) == 1 and orders[0].workflow == Pipeline.BALSAMIC
+
+
+@pytest.mark.parametrize(
+    "limit, expected_returned",
+    [(None, 2), (1, 1), (2, 2)],
+    ids=[
+        "Only workflow filtering",
+        "Workflow filtering and maximum one order",
+        "Workflow filtering and maximum two orders",
+    ],
+)
+def test_get_orders_mip_dna_and_limit_filter(
+    store: Store,
+    order: Order,
+    order_another: Order,
+    order_balsamic: Order,
+    limit: int | None,
+    expected_returned: int,
+):
+    # GIVEN a store with three orders, two of which are MIP-DNA orders
+
+    # WHEN fetching only MIP-DNA orders
+    orders: list[Order] = store.get_orders_by_workflow(workflow=Pipeline.MIP_DNA, limit=limit)
+
+    # THEN we should get the expected number of orders returned
+    assert len(orders) == expected_returned