(Streamline delivery) Add get order endpoint (#2898) (patch)

### Added

- Endpoint for fetching an order via order_id

cg/exc.py

@@ -252,3 +252,7 @@ class ArchiveJobFailedError(CgError):
 
 class XMLError(CgError):
     """Exception raised when something is wrong with the content of an XML file."""
+
+
+class OrderNotFoundError(CgError):
+    """Exception raised when an order is not found."""

cg/server/api.py

@@ -21,14 +21,20 @@
 from cg.apps.orderform.json_orderform_parser import JsonOrderformParser
 from cg.constants import ANALYSIS_SOURCES, METAGENOME_SOURCES
 from cg.constants.constants import FileFormat
-from cg.exc import CaseNotFoundError, OrderError, OrderFormError, TicketCreationError
+from cg.exc import (
+    CaseNotFoundError,
+    OrderError,
+    OrderFormError,
+    OrderNotFoundError,
+    TicketCreationError,
+)
 from cg.io.controller import WriteStream
 from cg.meta.orders import OrdersAPI
 from cg.models.orders.order import OrderIn, OrderType
 from cg.models.orders.orderform_schema import Orderform
 from cg.server.dto.delivery_message_response import DeliveryMessageResponse
 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, OrdersResponse
 from cg.server.ext import db, lims, osticket
 from cg.services.delivery_message.delivery_message_service import DeliveryMessageService
 from cg.services.orders.order_service import OrderService
@@ -479,6 +485,18 @@ def get_orders():
     return make_response(response.model_dump())
 
 
+@BLUEPRINT.route("/orders/<order_id>")
+def get_order(order_id: int):
+    """Return an order."""
+    order_service = OrderService(db)
+    try:
+        response: Order = order_service.get_order(order_id)
+        response_dict: dict = response.model_dump()
+        return make_response(response_dict)
+    except OrderNotFoundError as error:
+        return make_response(jsonify(error=str(error)), HTTPStatus.NOT_FOUND)
+
+
 @BLUEPRINT.route("/orderform", methods=["POST"])
 def parse_orderform():
     """Parse an orderform/JSON export."""

cg/services/orders/order_service.py

@@ -1,3 +1,4 @@
+from cg.exc import OrderNotFoundError
 from cg.models.orders.order import OrderIn
 from cg.server.dto.orders.orders_request import OrdersRequest
 from cg.server.dto.orders.orders_response import Order as OrderResponse
@@ -11,6 +12,12 @@ class OrderService:
     def __init__(self, store: Store) -> None:
         self.store = store
 
+    def get_order(self, order_id: int) -> OrderResponse:
+        order: Order | None = self.store.get_order_by_id(order_id)
+        if not order:
+            raise OrderNotFoundError(f"Order {order_id} not found.")
+        return create_order_response(order)
+
     def get_orders(self, orders_request: OrdersRequest) -> OrdersResponse:
         orders: list[Order] = self._get_orders(orders_request)
         return create_orders_response(orders)

cg/store/crud/read.py

@@ -1740,6 +1740,15 @@ def get_orders_by_workflow(
         )
         return orders.limit(limit).all()
 
+    def get_order_by_id(self, order_id: int) -> Order | None:
+        """Returns the entry in Order matching the given id."""
+        orders: Query = self._get_query(table=Order)
+        order_filter_functions: list[Callable] = [OrderFilter.FILTER_ORDERS_BY_ID]
+        orders: Query = apply_order_filters(
+            orders=orders, filter_functions=order_filter_functions, id=order_id
+        )
+        return orders.first()
+
     def _calculate_estimated_turnaround_time(
         self,
         is_rerun,

cg/store/filters/status_order_filters.py

@@ -11,14 +11,22 @@ def filter_orders_by_workflow(orders: Query, workflow: str, **kwargs) -> Query:
     return orders.filter(Order.workflow == workflow) if workflow else orders
 
 
-def apply_order_filters(filter_functions: list[Callable], orders: Query, workflow: str) -> Query:
+def filter_orders_by_id(orders: Query, id: int, **kwargs) -> Query:
+    """Return orders filtered on id."""
+    return orders.filter(Order.id == id)
+
+
+def apply_order_filters(
+    filter_functions: list[Callable], orders: Query, id: int = None, workflow: str = None
+) -> 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)
+        orders: Query = filter_function(orders=orders, id=id, workflow=workflow)
     return orders
 
 
 class OrderFilter(Enum):
     """Define order filter functions."""
 
+    FILTER_ORDERS_BY_ID: Callable = filter_orders_by_id
     FILTER_ORDERS_BY_WORKFLOW: Callable = filter_orders_by_workflow

tests/server/conftest.py

@@ -90,6 +90,11 @@ def order_balsamic(helpers: StoreHelpers, customer_another: Customer) -> Order:
     return order
 
 
+@pytest.fixture
+def non_existent_order_id() -> int:
+    return 900
+
+
 @pytest.fixture
 def client(app: Flask) -> Generator[FlaskClient, None, None]:
     # Bypass authentication

tests/server/endpoints/test_orders_endpoint.py

@@ -4,6 +4,7 @@
 from flask.testing import FlaskClient
 
 from cg.constants import Workflow
+from cg.services.orders.utils import create_order_response
 from cg.store.models import Order
 
 
@@ -39,3 +40,38 @@ def test_orders_endpoint(
 
     # THEN the response contains the correct number of orders
     assert len(response.json["orders"]) == expected_orders
+
+
+def test_order_endpoint(
+    client: FlaskClient,
+    order: Order,
+    order_another: Order,
+):
+    """Tests that the order endpoint returns the order with matching id"""
+    # GIVEN a store with two orders
+
+    order_id_to_fetch: int = order.id
+
+    # WHEN a request is made to get a specific order
+    endpoint: str = f"/api/v1/orders/{order_id_to_fetch}"
+    response = client.get(endpoint)
+
+    # THEN the response should be successful
+    assert response.status_code == HTTPStatus.OK
+
+    # THEN the response should only contain the specified order
+    assert response.json == create_order_response(order).model_dump()
+
+
+def test_order_endpoint_not_found(
+    client: FlaskClient, order: Order, order_another: Order, non_existent_order_id: int
+):
+    """Tests that the order endpoint returns the order with matching id"""
+    # GIVEN a store with two orders
+
+    # WHEN a request is made to get a non-existent order
+    endpoint: str = f"/api/v1/orders/{non_existent_order_id}"
+    response = client.get(endpoint)
+
+    # THEN the response should be unsuccessful
+    assert response.status_code == HTTPStatus.NOT_FOUND