12596434604

Committed 03 Jan 2025 10:37AM UTC coverage: 87.616%. First build

Build # 12596434604

Build Type

Pull #507

github

Committed by

ChrOertlin

Commit Message

fix: fix error

Pull Request Pull Request #507: feat( user token verification)

Run Details

25 of 52 new or added lines in 4 files covered. (48.08%)

2080 of 2374 relevant lines covered (87.62%)

0.88 hits per line

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

73.33

/trailblazer/server/api.py

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 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.server.ext import store
from trailblazer.server.utils import (
    handle_endpoint_errors,
    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.job_service import JobService
from trailblazer.services.user_verification_service.exc import UserTokenVerificationError
from trailblazer.services.user_verification_service.user_verification_service import (
    UserVerificationService,
)
from trailblazer.store.models import Info, User

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


@blueprint.before_request
@inject
def before_request(
    user_verification_service: UserVerificationService = Provide[
        Container.user_verification_service
    ],
):
    """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
    try:
        g.current_user = user_verification_service.verify_user(request.headers.get("Authorization"))
    except (UserTokenVerificationError, ValueError) as error:
        abort(HTTPStatus.UNAUTHORIZED, str(error))


@blueprint.route("/auth", methods=["POST"])
@handle_endpoint_errors
@inject
def authenticate(auth_service: AuthenticationService = Provide[Container.auth_service]):
    """Exchange authorization code for an access token."""
    request_data = CodeExchangeRequest.model_validate(request.json)
    token: str = auth_service.authenticate(request_data.code)
    return jsonify({"token": token}), HTTPStatus.OK


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


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


@blueprint.route("/analyses", methods=["PATCH"])
@handle_endpoint_errors
@inject
def patch_analyses(analysis_service: AnalysisService = Provide[Container.analysis_service]):
    """Update data (such as status, visibility, comments etc.) for multiple analyses at once."""
    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


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


@blueprint.route("/analyses/<int:analysis_id>/cancel", methods=["POST"])
@handle_endpoint_errors
@inject
def cancel_analysis(
    analysis_id: int,
    analysis_service: AnalysisService = Provide[Container.analysis_service],
):
    response: AnalysisResponse = analysis_service.cancel_analysis_from_web(analysis_id)
    return jsonify(response.model_dump()), HTTPStatus.OK


@blueprint.route("/analysis/<int:analysis_id>/jobs", methods=["POST"])
@handle_endpoint_errors
@inject
def add_job(analysis_id: int, job_service: JobService = Provide[Container.job_service]):
    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


@blueprint.route("/analyses/<int:analysis_id>", methods=["PUT"])
@handle_endpoint_errors
@inject
def update_analysis(
    analysis_id: int, analysis_service: AnalysisService = Provide[Container.analysis_service]
):
    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


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


@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"])
@handle_endpoint_errors
@inject
def get_failed_jobs(job_service: JobService = Provide[Container.job_service]):
    request_data = FailedJobsRequest.model_validate(request.args)
    response: FailedJobsResponse = job_service.get_failed_jobs(request_data)
    return jsonify(response.model_dump()), HTTPStatus.OK


# 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"])
@handle_endpoint_errors
@inject
def add_pending_analysis(analysis_service: AnalysisService = Provide[Container.analysis_service]):
    request_data = CreateAnalysisRequest.model_validate(request.json)
    response: AnalysisResponse = analysis_service.add_pending_analysis(request_data)
    return jsonify(response.model_dump()), HTTPStatus.CREATED


@blueprint.route("/set-analysis-uploaded", methods=["PUT"])
@handle_endpoint_errors
def set_analysis_uploaded():
    """Set the analysis uploaded at attribute."""
    put_request: Response.json = request.json
    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


@blueprint.route("/set-analysis-status", methods=["PUT"])
@handle_endpoint_errors
def set_analysis_status():
    """Update analysis status of a case with supplied status."""
    put_request: Response.json = request.json
    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,
    )


@blueprint.route("/add-comment", methods=["PUT"])
@handle_endpoint_errors
def add_comment():
    """Updating comment on analysis."""
    put_request: Response.json = request.json
    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

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

Clinical-Genomics / trailblazer / 12596434604

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