9302500416

Committed 30 May 2024 12:27PM UTC coverage: 90.364%. First build

Build # 9302500416

Build Type

Pull #449

github

Committed by

islean

Commit Message

Refactor

Pull Request Pull Request #449: Add to comment instead of exchanging

Run Details

6 of 7 new or added lines in 2 files covered. (85.71%)

2232 of 2470 relevant lines covered (90.36%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.79

/trailblazer/server/api.py

import logging
import os
from http import HTTPStatus
from typing import Mapping

from dependency_injector.wiring import Provide, inject
from flask import Blueprint, Response, abort, g, jsonify, make_response, request
from google.auth import jwt
from pydantic import ValidationError

from trailblazer.containers import Container
from trailblazer.dto import (
    AnalysesRequest,
    AnalysesResponse,
    AnalysisResponse,
    AnalysisUpdateRequest,
    CreateJobRequest,
    FailedJobsRequest,
    FailedJobsResponse,
    JobResponse,
)
from trailblazer.dto.analyses_response import UpdateAnalysesResponse
from trailblazer.dto.authentication.code_exchange_request import CodeExchangeRequest
from trailblazer.dto.create_analysis_request import CreateAnalysisRequest
from trailblazer.dto.summaries_request import SummariesRequest
from trailblazer.dto.summaries_response import SummariesResponse
from trailblazer.dto.update_analyses import UpdateAnalyses
from trailblazer.exc import MissingAnalysis
from trailblazer.server.ext import store
from trailblazer.server.utils import parse_analyses_request, stringify_timestamps
from trailblazer.services.analysis_service.analysis_service import AnalysisService
from trailblazer.services.authentication_service.authentication_service import AuthenticationService
from trailblazer.services.authentication_service.exceptions import AuthenticationError
from trailblazer.services.job_service import JobService
from trailblazer.store.models import Info, User

blueprint = Blueprint("api", __name__, url_prefix="/api/v1")


@blueprint.before_request
def before_request():
    """Authentication that is run before processing requests to the application"""
    if request.endpoint == "api.authenticate":
        return
    if request.method == "OPTIONS":
        return make_response(jsonify(ok=True), 204)
    if os.environ.get("SCOPE") == "DEVELOPMENT":
        return
    if auth_header := request.headers.get("Authorization"):
        jwt_token = auth_header.split("Bearer ")[-1]
    else:
        return abort(403, "no JWT token found on request")

    user_data: Mapping = jwt.decode(jwt_token, verify=False)
    if user := store.get_user(email=user_data["email"], exclude_archived=True):
        g.current_user = user
    else:
        return abort(403, f"{user_data['email']} doesn't have access")


@blueprint.route("/auth", methods=["POST"])
@inject
def authenticate(auth_service: AuthenticationService = Provide[Container.auth_service]):
    """Exchange authorization code for an access token."""
    try:
        request_data = CodeExchangeRequest.model_validate(request.json)
        token: str = auth_service.authenticate(request_data.code)
        return jsonify({"token": token}), HTTPStatus.OK
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST
    except AuthenticationError:
        return jsonify("User not allowed"), HTTPStatus.FORBIDDEN


@blueprint.route("/auth/refresh", methods=["GET"])
@inject
def refresh_token(auth_service: AuthenticationService = Provide[Container.auth_service]):
    """Refresh access token."""
    user: User = g.current_user
    try:
        token: str = auth_service.refresh_token(user.id)
        return jsonify({"token": token}), HTTPStatus.OK
    except AuthenticationError:
        return jsonify("User not allowed"), HTTPStatus.FORBIDDEN


@blueprint.route("/analyses", methods=["GET"])
@inject
def get_analyses(analysis_service: AnalysisService = Provide[Container.analysis_service]):
    try:
        request_data: AnalysesRequest = parse_analyses_request(request)
        response: AnalysesResponse = analysis_service.get_analyses(request_data)
        return jsonify(response.model_dump()), HTTPStatus.OK
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST


@blueprint.route("/analyses", methods=["PATCH"])
@inject
def patch_analyses(analysis_service: AnalysisService = Provide[Container.analysis_service]):
    """Update data (such as status, visibility, comments etc.) for multiple analyses at once."""
    try:
        request_data = UpdateAnalyses.model_validate(request.json)
        user: User = g.get("current_user")
        response: UpdateAnalysesResponse = analysis_service.update_analyses(
            data=request_data, user=user
        )
        return jsonify(response.model_dump()), HTTPStatus.OK
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST
    except MissingAnalysis as error:
        return jsonify(error=str(error)), HTTPStatus.NOT_FOUND


@blueprint.route("/analyses/<int:analysis_id>", methods=["GET"])
@inject
def get_analysis(
    analysis_id: int, analysis_service: AnalysisService = Provide[Container.analysis_service]
):
    try:
        response: AnalysisResponse = analysis_service.get_analysis(analysis_id)
        return jsonify(response.model_dump()), HTTPStatus.OK
    except MissingAnalysis as error:
        return jsonify(error=str(error)), HTTPStatus.NOT_FOUND


@blueprint.route("/analysis/<int:analysis_id>/jobs", methods=["POST"])
@inject
def add_job(analysis_id: int, job_service: JobService = Provide[Container.job_service]):
    try:
        data = CreateJobRequest.model_validate(request.json)
        response: JobResponse = job_service.add_job(analysis_id=analysis_id, data=data)
        return jsonify(response.model_dump()), HTTPStatus.CREATED
    except MissingAnalysis as error:
        return jsonify(error=str(error)), HTTPStatus.NOT_FOUND
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST


@blueprint.route("/analyses/<int:analysis_id>", methods=["PUT"])
@inject
def update_analysis(
    analysis_id: int, analysis_service: AnalysisService = Provide[Container.analysis_service]
):
    try:
        request_data = AnalysisUpdateRequest.model_validate(request.json)
        user: User = g.get("current_user")
        response: AnalysisResponse = analysis_service.update_analysis(
            analysis_id=analysis_id, update=request_data, user=user
        )
        return jsonify(response.model_dump()), HTTPStatus.OK
    except MissingAnalysis as error:
        return jsonify(error=str(error)), HTTPStatus.NOT_FOUND
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST


@blueprint.route("/summary", methods=["GET"])
@inject
def get_summaries(analysis_service: AnalysisService = Provide[Container.analysis_service]):
    try:
        request_data = SummariesRequest.model_validate(request.args)
        response: SummariesResponse = analysis_service.get_summaries(request_data)
        return jsonify(response.model_dump()), HTTPStatus.OK
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST


@blueprint.route("/info")
def info():
    """Display metadata about database."""
    info: Info = store.get_query(table=Info).first()
    return jsonify(**info.to_dict())


@blueprint.route("/me")
def me():
    """Return information about a logged in user."""
    return jsonify(**g.current_user.to_dict())


@blueprint.route("/aggregate/jobs", methods=["GET"])
@inject
def get_failed_jobs(job_service: JobService = Provide[Container.job_service]):
    try:
        request_data = FailedJobsRequest.model_validate(request.args)
        response: FailedJobsResponse = job_service.get_failed_jobs(request_data)
        return jsonify(response.model_dump()), HTTPStatus.OK
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST


# CG REST INTERFACE ###
# ONLY POST routes which accept messages in specific format
# NOT for use with GUI (for now)


@blueprint.route("/get-latest-analysis", methods=["POST"])
def post_get_latest_analysis():
    """Return latest analysis entry for specified case id."""
    post_request = request.json
    case_id: str = post_request.get("case_id")
    if latest_case_analysis := store.get_latest_analysis_for_case(case_id):
        raw_analysis: dict[str, str] = stringify_timestamps(latest_case_analysis.to_dict())
        return jsonify(**raw_analysis), HTTPStatus.OK
    return jsonify(None), HTTPStatus.OK


@blueprint.route("/add-pending-analysis", methods=["POST"])
@inject
def post_add_pending_analysis(
    analysis_service: AnalysisService = Provide[Container.analysis_service],
):
    try:
        request_data = CreateAnalysisRequest.model_validate(request.json)
        response: AnalysisResponse = analysis_service.add_pending_analysis(request_data)
        return jsonify(response.model_dump()), HTTPStatus.CREATED
    except ValidationError as error:
        return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST


@blueprint.route("/set-analysis-uploaded", methods=["PUT"])
def set_analysis_uploaded():
    """Set the analysis uploaded at attribute."""
    put_request: Response.json = request.json
    try:
        store.update_analysis_uploaded_at(
            case_id=put_request.get("case_id"), uploaded_at=put_request.get("uploaded_at")
        )
        return jsonify("Success! Uploaded at request sent"), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT


@blueprint.route("/set-analysis-status", methods=["PUT"])
def set_analysis_status():
    """Update analysis status of a case with supplied status."""
    put_request: Response.json = request.json
    try:
        case_id: str = put_request.get("case_id")
        status: str = put_request.get("status")
        store.update_analysis_status_by_case_id(case_id=case_id, status=status)
        return (
            jsonify(f"Success! Analysis set to {put_request.get('status')} request sent"),
            HTTPStatus.CREATED,
        )
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT


@blueprint.route("/add-comment", methods=["PUT"])
def add_comment():
    """Updating comment on analysis."""
    put_request: Response.json = request.json
    try:
        case_id: str = put_request.get("case_id")
        comment: str = put_request.get("comment")
        store.update_latest_analysis_comment(case_id=case_id, comment=comment)
        return jsonify("Success! Adding comment request sent"), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT

1	import logging	1✔
2	import os	1✔
3	from http import HTTPStatus	1✔
4	from typing import Mapping	1✔
5
6	from dependency_injector.wiring import Provide, inject	1✔
7	from flask import Blueprint, Response, abort, g, jsonify, make_response, request	1✔
8	from google.auth import jwt	1✔
9	from pydantic import ValidationError	1✔
10
11	from trailblazer.containers import Container	1✔
12	from trailblazer.dto import (	1✔
13	AnalysesRequest,
14	AnalysesResponse,
15	AnalysisResponse,
16	AnalysisUpdateRequest,
17	CreateJobRequest,
18	FailedJobsRequest,
19	FailedJobsResponse,
20	JobResponse,
21	)
22	from trailblazer.dto.analyses_response import UpdateAnalysesResponse	1✔
23	from trailblazer.dto.authentication.code_exchange_request import CodeExchangeRequest	1✔
24	from trailblazer.dto.create_analysis_request import CreateAnalysisRequest	1✔
25	from trailblazer.dto.summaries_request import SummariesRequest	1✔
26	from trailblazer.dto.summaries_response import SummariesResponse	1✔
27	from trailblazer.dto.update_analyses import UpdateAnalyses	1✔
28	from trailblazer.exc import MissingAnalysis	1✔
29	from trailblazer.server.ext import store	1✔
30	from trailblazer.server.utils import parse_analyses_request, stringify_timestamps	1✔
31	from trailblazer.services.analysis_service.analysis_service import AnalysisService	1✔
32	from trailblazer.services.authentication_service.authentication_service import AuthenticationService	1✔
33	from trailblazer.services.authentication_service.exceptions import AuthenticationError	1✔
34	from trailblazer.services.job_service import JobService	1✔
35	from trailblazer.store.models import Info, User	1✔
36
37	blueprint = Blueprint("api", __name__, url_prefix="/api/v1")	1✔
38
39
40	@blueprint.before_request	1✔
41	def before_request():	1✔
42	"""Authentication that is run before processing requests to the application"""
43	if request.endpoint == "api.authenticate":	×
44	return	×
45	if request.method == "OPTIONS":	×
46	return make_response(jsonify(ok=True), 204)	×
47	if os.environ.get("SCOPE") == "DEVELOPMENT":	×
48	return	×
49	if auth_header := request.headers.get("Authorization"):	×
50	jwt_token = auth_header.split("Bearer ")[-1]	×
51	else:
52	return abort(403, "no JWT token found on request")	×
53
54	user_data: Mapping = jwt.decode(jwt_token, verify=False)	×
55	if user := store.get_user(email=user_data["email"], exclude_archived=True):	×
56	g.current_user = user	×
57	else:
58	return abort(403, f"{user_data['email']} doesn't have access")	×
59
60
61	@blueprint.route("/auth", methods=["POST"])	1✔
62	@inject	1✔
63	def authenticate(auth_service: AuthenticationService = Provide[Container.auth_service]):	1✔
64	"""Exchange authorization code for an access token."""
65	try:	×
66	request_data = CodeExchangeRequest.model_validate(request.json)	×
67	token: str = auth_service.authenticate(request_data.code)	×
68	return jsonify({"token": token}), HTTPStatus.OK	×
69	except ValidationError as error:	×
70	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
71	except AuthenticationError:	×
72	return jsonify("User not allowed"), HTTPStatus.FORBIDDEN	×
73
74
75	@blueprint.route("/auth/refresh", methods=["GET"])	1✔
76	@inject	1✔
77	def refresh_token(auth_service: AuthenticationService = Provide[Container.auth_service]):	1✔
78	"""Refresh access token."""
79	user: User = g.current_user	×
80	try:	×
81	token: str = auth_service.refresh_token(user.id)	×
82	return jsonify({"token": token}), HTTPStatus.OK	×
83	except AuthenticationError:	×
84	return jsonify("User not allowed"), HTTPStatus.FORBIDDEN	×
85
86
87	@blueprint.route("/analyses", methods=["GET"])	1✔
88	@inject	1✔
89	def get_analyses(analysis_service: AnalysisService = Provide[Container.analysis_service]):	1✔
90	try:	1✔
91	request_data: AnalysesRequest = parse_analyses_request(request)	1✔
92	response: AnalysesResponse = analysis_service.get_analyses(request_data)	1✔
93	return jsonify(response.model_dump()), HTTPStatus.OK	1✔
94	except ValidationError as error:	×
95	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
96
97
98	@blueprint.route("/analyses", methods=["PATCH"])	1✔
99	@inject	1✔
100	def patch_analyses(analysis_service: AnalysisService = Provide[Container.analysis_service]):	1✔
101	"""Update data (such as status, visibility, comments etc.) for multiple analyses at once."""
102	try:	1✔
103	request_data = UpdateAnalyses.model_validate(request.json)	1✔
104	user: User = g.get("current_user")	1✔
105	response: UpdateAnalysesResponse = analysis_service.update_analyses(	1✔
106	data=request_data, user=user
107	)
108	return jsonify(response.model_dump()), HTTPStatus.OK	1✔
109	except ValidationError as error:	×
110	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
111	except MissingAnalysis as error:	×
112	return jsonify(error=str(error)), HTTPStatus.NOT_FOUND	×
113
114
115	@blueprint.route("/analyses/<int:analysis_id>", methods=["GET"])	1✔
116	@inject	1✔
117	def get_analysis(	1✔
118	analysis_id: int, analysis_service: AnalysisService = Provide[Container.analysis_service]
119	):
120	try:	1✔
121	response: AnalysisResponse = analysis_service.get_analysis(analysis_id)	1✔
122	return jsonify(response.model_dump()), HTTPStatus.OK	1✔
123	except MissingAnalysis as error:	1✔
124	return jsonify(error=str(error)), HTTPStatus.NOT_FOUND	1✔
125
126
127	@blueprint.route("/analysis/<int:analysis_id>/jobs", methods=["POST"])	1✔
128	@inject	1✔
129	def add_job(analysis_id: int, job_service: JobService = Provide[Container.job_service]):	1✔
130	try:	1✔
131	data = CreateJobRequest.model_validate(request.json)	1✔
132	response: JobResponse = job_service.add_job(analysis_id=analysis_id, data=data)	1✔
133	return jsonify(response.model_dump()), HTTPStatus.CREATED	1✔
134	except MissingAnalysis as error:	×
135	return jsonify(error=str(error)), HTTPStatus.NOT_FOUND	×
136	except ValidationError as error:	×
137	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
138
139
140	@blueprint.route("/analyses/<int:analysis_id>", methods=["PUT"])	1✔
141	@inject	1✔
142	def update_analysis(	1✔
143	analysis_id: int, analysis_service: AnalysisService = Provide[Container.analysis_service]
144	):
145	try:	1✔
146	request_data = AnalysisUpdateRequest.model_validate(request.json)	1✔
147	user: User = g.get("current_user")	1✔
148	response: AnalysisResponse = analysis_service.update_analysis(	1✔
149	analysis_id=analysis_id, update=request_data, user=user
150	)
151	return jsonify(response.model_dump()), HTTPStatus.OK	1✔
152	except MissingAnalysis as error:	1✔
153	return jsonify(error=str(error)), HTTPStatus.NOT_FOUND	×
154	except ValidationError as error:	1✔
155	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	1✔
156
157
158	@blueprint.route("/summary", methods=["GET"])	1✔
159	@inject	1✔
160	def get_summaries(analysis_service: AnalysisService = Provide[Container.analysis_service]):	1✔
161	try:	1✔
162	request_data = SummariesRequest.model_validate(request.args)	1✔
163	response: SummariesResponse = analysis_service.get_summaries(request_data)	1✔
164	return jsonify(response.model_dump()), HTTPStatus.OK	1✔
165	except ValidationError as error:	×
166	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
167
168
169	@blueprint.route("/info")	1✔
170	def info():	1✔
171	"""Display metadata about database."""
172	info: Info = store.get_query(table=Info).first()	×
173	return jsonify(**info.to_dict())	×
174
175
176	@blueprint.route("/me")	1✔
177	def me():	1✔
178	"""Return information about a logged in user."""
179	return jsonify(**g.current_user.to_dict())	×
180
181
182	@blueprint.route("/aggregate/jobs", methods=["GET"])	1✔
183	@inject	1✔
184	def get_failed_jobs(job_service: JobService = Provide[Container.job_service]):	1✔
185	try:	1✔
186	request_data = FailedJobsRequest.model_validate(request.args)	1✔
187	response: FailedJobsResponse = job_service.get_failed_jobs(request_data)	1✔
188	return jsonify(response.model_dump()), HTTPStatus.OK	1✔
189	except ValidationError as error:	×
190	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
191
192
193	# CG REST INTERFACE ###
194	# ONLY POST routes which accept messages in specific format
195	# NOT for use with GUI (for now)
196
197
198	@blueprint.route("/get-latest-analysis", methods=["POST"])	1✔
199	def post_get_latest_analysis():	1✔
200	"""Return latest analysis entry for specified case id."""
201	post_request = request.json	×
202	case_id: str = post_request.get("case_id")	×
203	if latest_case_analysis := store.get_latest_analysis_for_case(case_id):	×
204	raw_analysis: dict[str, str] = stringify_timestamps(latest_case_analysis.to_dict())	×
205	return jsonify(**raw_analysis), HTTPStatus.OK	×
206	return jsonify(None), HTTPStatus.OK	×
207
208
209	@blueprint.route("/add-pending-analysis", methods=["POST"])	1✔
210	@inject	1✔
211	def post_add_pending_analysis(	1✔
212	analysis_service: AnalysisService = Provide[Container.analysis_service],
213	):
214	try:	1✔
215	request_data = CreateAnalysisRequest.model_validate(request.json)	1✔
216	response: AnalysisResponse = analysis_service.add_pending_analysis(request_data)	1✔
217	return jsonify(response.model_dump()), HTTPStatus.CREATED	1✔
218	except ValidationError as error:	×
219	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
220
221
222	@blueprint.route("/set-analysis-uploaded", methods=["PUT"])	1✔
223	def set_analysis_uploaded():	1✔
224	"""Set the analysis uploaded at attribute."""
225	put_request: Response.json = request.json	×
226	try:	×
227	store.update_analysis_uploaded_at(	×
228	case_id=put_request.get("case_id"), uploaded_at=put_request.get("uploaded_at")
229	)
230	return jsonify("Success! Uploaded at request sent"), HTTPStatus.CREATED	×
231	except Exception as error:	×
232	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
233
234
235	@blueprint.route("/set-analysis-status", methods=["PUT"])	1✔
236	def set_analysis_status():	1✔
237	"""Update analysis status of a case with supplied status."""
238	put_request: Response.json = request.json	×
239	try:	×
240	case_id: str = put_request.get("case_id")	×
241	status: str = put_request.get("status")	×
242	store.update_analysis_status_by_case_id(case_id=case_id, status=status)	×
243	return (	×
244	jsonify(f"Success! Analysis set to {put_request.get('status')} request sent"),
245	HTTPStatus.CREATED,
246	)
247	except Exception as error:	×
248	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
249
250
251	@blueprint.route("/add-comment", methods=["PUT"])	1✔
252	def add_comment():	1✔
253	"""Updating comment on analysis."""
254	put_request: Response.json = request.json	×
255	try:	×
256	case_id: str = put_request.get("case_id")	×
257	comment: str = put_request.get("comment")	×
NEW 258	store.update_latest_analysis_comment(case_id=case_id, comment=comment)	×
259	return jsonify("Success! Adding comment request sent"), HTTPStatus.CREATED	×
260	except Exception as error:	×
261	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×

Clinical-Genomics / trailblazer / 9302500416

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous