From 15964900527053b2c5b95e1b5c7477599cab310b Mon Sep 17 00:00:00 2001 From: Sebastian Allard Date: Thu, 8 Aug 2024 14:56:07 +0200 Subject: [PATCH] Extract case endpoints (#3515) --- cg/server/api.py | 100 +------------------------------- cg/server/app.py | 3 + cg/server/endpoints/cases.py | 108 +++++++++++++++++++++++++++++++++++ 3 files changed, 113 insertions(+), 98 deletions(-) create mode 100644 cg/server/endpoints/cases.py diff --git a/cg/server/api.py b/cg/server/api.py index 529fd624bd..76673b4339 100644 --- a/cg/server/api.py +++ b/cg/server/api.py @@ -17,11 +17,9 @@ from cg.constants import ANALYSIS_SOURCES, METAGENOME_SOURCES from cg.constants.constants import FileFormat from cg.exc import ( - CaseNotFoundError, OrderError, OrderExistsError, OrderFormError, - OrderMismatchError, OrderNotDeliverableError, OrderNotFoundError, TicketCreationError, @@ -31,6 +29,8 @@ from cg.meta.orders.ticket_handler import TicketHandler from cg.models.orders.order import OrderIn, OrderType from cg.models.orders.orderform_schema import Orderform +from cg.server.dto.delivery_message.delivery_message_response import DeliveryMessageResponse +from cg.server.dto.orders.order_delivery_update_request import OrderDeliveredUpdateRequest from cg.server.dto.delivery_message.delivery_message_request import ( DeliveryMessageRequest, ) @@ -53,7 +53,6 @@ Analysis, Application, ApplicationLimitations, - Case, Customer, IlluminaSampleSequencingMetrics, Pool, @@ -126,101 +125,6 @@ def submit_order(order_type): return abort(make_response(jsonify(message=error_message), http_error_response)) -@BLUEPRINT.route("/cases") -def get_cases(): - """Return cases with links for a customer from the database.""" - enquiry: str = request.args.get("enquiry") - action: str = request.args.get("action") - - customers: list[Customer] = _get_current_customers() - cases: list[Case] = _get_cases(enquiry=enquiry, action=action, customers=customers) - - nr_cases: int = len(cases) - cases_with_links: list[dict] = [case.to_dict(links=True) for case in cases] - return jsonify(families=cases_with_links, total=nr_cases) - - -def _get_current_customers() -> list[Customer] | None: - """Return customers if the current user is not an admin.""" - return g.current_user.customers if not g.current_user.is_admin else None - - -def _get_cases( - enquiry: str | None, action: str | None, customers: list[Customer] | None -) -> list[Case]: - """Get cases based on the provided filters.""" - return db.get_cases_by_customers_action_and_case_search( - case_search=enquiry, - customers=customers, - action=action, - ) - - -@BLUEPRINT.route("/cases/") -def parse_case(case_id): - """Return a case with links.""" - case: Case = db.get_case_by_internal_id(internal_id=case_id) - if case is None: - return abort(HTTPStatus.NOT_FOUND) - if not g.current_user.is_admin and (case.customer not in g.current_user.customers): - return abort(HTTPStatus.FORBIDDEN) - return jsonify(**case.to_dict(links=True, analyses=True)) - - -@BLUEPRINT.route("/cases/delivery_message", methods=["GET"]) -def get_cases_delivery_message(): - delivery_message_request = DeliveryMessageRequest.model_validate(request.args) - try: - response: DeliveryMessageResponse = delivery_message_service.get_cases_message( - delivery_message_request - ) - return jsonify(response.model_dump()), HTTPStatus.OK - except (CaseNotFoundError, OrderMismatchError) as error: - return jsonify({"error": str(error)}), HTTPStatus.BAD_REQUEST - - -@BLUEPRINT.route("/cases//delivery_message", methods=["GET"]) -def get_case_delivery_message(case_id: str): - delivery_message_request = DeliveryMessageRequest(case_ids=[case_id]) - try: - response: DeliveryMessageResponse = delivery_message_service.get_cases_message( - delivery_message_request - ) - return jsonify(response.model_dump()), HTTPStatus.OK - except CaseNotFoundError as error: - return jsonify({"error": str(error)}), HTTPStatus.BAD_REQUEST - - -@BLUEPRINT.route("/families_in_collaboration") -def parse_families_in_collaboration(): - """Return cases in collaboration.""" - - customer_internal_id = request.args.get("customer") - workflow = request.args.get("data_analysis") - case_search_pattern = request.args.get("enquiry") - - customer = db.get_customer_by_internal_id(customer_internal_id=customer_internal_id) - - cases = db.get_cases_by_customer_workflow_and_case_search( - customer=customer, workflow=workflow, case_search=case_search_pattern - ) - - case_dicts = [case.to_dict(links=True) for case in cases] - return jsonify(families=case_dicts, total=len(cases)) - - -@BLUEPRINT.route("/families_in_collaboration/") -def parse_family_in_collaboration(family_id): - """Return a family with links.""" - case: Case = db.get_case_by_internal_id(internal_id=family_id) - customer: Customer = db.get_customer_by_internal_id( - customer_internal_id=request.args.get("customer") - ) - if case.customer not in customer.collaborators: - return abort(HTTPStatus.FORBIDDEN) - return jsonify(**case.to_dict(links=True, analyses=True)) - - @BLUEPRINT.route("/samples") def parse_samples(): """Return samples.""" diff --git a/cg/server/app.py b/cg/server/app.py index 9d68690fa3..d96e837957 100644 --- a/cg/server/app.py +++ b/cg/server/app.py @@ -8,6 +8,7 @@ from cg.server import admin, api, ext, invoices from cg.server.app_config import app_config +from cg.server.endpoints.cases import CASES_BLUEPRINT from cg.server.endpoints.samples import SAMPLES_BLUEPRINT from cg.store.database import get_scoped_session_registry from cg.store.models import ( @@ -88,9 +89,11 @@ def logged_in(blueprint, token): app.register_blueprint(invoices.BLUEPRINT, url_prefix="/invoices") app.register_blueprint(oauth_bp, url_prefix="/login") app.register_blueprint(SAMPLES_BLUEPRINT) + app.register_blueprint(CASES_BLUEPRINT) _register_admin_views() ext.csrf.exempt(api.BLUEPRINT) # Protected with Auth header already + ext.csrf.exempt(CASES_BLUEPRINT) # Protected with Auth header already @app.route("/") def index(): diff --git a/cg/server/endpoints/cases.py b/cg/server/endpoints/cases.py new file mode 100644 index 0000000000..4d03f47c43 --- /dev/null +++ b/cg/server/endpoints/cases.py @@ -0,0 +1,108 @@ +import logging +from http import HTTPStatus +from flask import Blueprint, abort, g, jsonify, request +from cg.exc import CaseNotFoundError, OrderMismatchError +from cg.server.dto.delivery_message.delivery_message_request import DeliveryMessageRequest +from cg.server.dto.delivery_message.delivery_message_response import DeliveryMessageResponse +from cg.server.endpoints.utils import before_request +from cg.server.ext import db, delivery_message_service +from cg.store.models import Case, Customer + +LOG = logging.getLogger(__name__) +CASES_BLUEPRINT = Blueprint("cases", __name__, url_prefix="/api/v1") +CASES_BLUEPRINT.before_request(before_request) + + +@CASES_BLUEPRINT.route("/cases") +def get_cases(): + """Return cases with links for a customer from the database.""" + enquiry: str = request.args.get("enquiry") + action: str = request.args.get("action") + + customers: list[Customer] = _get_current_customers() + cases: list[Case] = _get_cases(enquiry=enquiry, action=action, customers=customers) + + nr_cases: int = len(cases) + cases_with_links: list[dict] = [case.to_dict(links=True) for case in cases] + return jsonify(families=cases_with_links, total=nr_cases) + + +def _get_current_customers() -> list[Customer] | None: + """Return customers if the current user is not an admin.""" + return g.current_user.customers if not g.current_user.is_admin else None + + +def _get_cases( + enquiry: str | None, action: str | None, customers: list[Customer] | None +) -> list[Case]: + """Get cases based on the provided filters.""" + return db.get_cases_by_customers_action_and_case_search( + case_search=enquiry, + customers=customers, + action=action, + ) + + +@CASES_BLUEPRINT.route("/cases/") +def parse_case(case_id): + """Return a case with links.""" + case: Case = db.get_case_by_internal_id(internal_id=case_id) + if case is None: + return abort(HTTPStatus.NOT_FOUND) + if not g.current_user.is_admin and (case.customer not in g.current_user.customers): + return abort(HTTPStatus.FORBIDDEN) + return jsonify(**case.to_dict(links=True, analyses=True)) + + +@CASES_BLUEPRINT.route("/cases/delivery_message", methods=["GET"]) +def get_cases_delivery_message(): + delivery_message_request = DeliveryMessageRequest.model_validate(request.args) + try: + response: DeliveryMessageResponse = delivery_message_service.get_cases_message( + delivery_message_request + ) + return jsonify(response.model_dump()), HTTPStatus.OK + except (CaseNotFoundError, OrderMismatchError) as error: + return jsonify({"error": str(error)}), HTTPStatus.BAD_REQUEST + + +@CASES_BLUEPRINT.route("/cases//delivery_message", methods=["GET"]) +def get_case_delivery_message(case_id: str): + delivery_message_request = DeliveryMessageRequest(case_ids=[case_id]) + try: + response: DeliveryMessageResponse = delivery_message_service.get_cases_message( + delivery_message_request + ) + return jsonify(response.model_dump()), HTTPStatus.OK + except CaseNotFoundError as error: + return jsonify({"error": str(error)}), HTTPStatus.BAD_REQUEST + + +@CASES_BLUEPRINT.route("/families_in_collaboration") +def get_cases_in_collaboration(): + """Return cases in collaboration.""" + + customer_internal_id = request.args.get("customer") + workflow = request.args.get("data_analysis") + case_search_pattern = request.args.get("enquiry") + + customer = db.get_customer_by_internal_id(customer_internal_id=customer_internal_id) + + cases = db.get_cases_by_customer_workflow_and_case_search( + customer=customer, workflow=workflow, case_search=case_search_pattern + ) + + case_dicts = [case.to_dict(links=True) for case in cases] + return jsonify(families=case_dicts, total=len(cases)) + + +@CASES_BLUEPRINT.route("/families_in_collaboration/") +def get_case_in_collaboration(family_id): + """Return a case with links.""" + case: Case = db.get_case_by_internal_id(internal_id=family_id) + customer: Customer = db.get_customer_by_internal_id( + customer_internal_id=request.args.get("customer") + ) + if case.customer not in customer.collaborators: + return abort(HTTPStatus.FORBIDDEN) + return jsonify(**case.to_dict(links=True, analyses=True))