7005273242

Committed 27 Nov 2023 12:54PM UTC coverage: 77.88%. First build

Build # 7005273242

Build Type

Pull #319

github

Committed by

seallard

Commit Message

Fix type hints

Pull Request Pull Request #319: Patch update endpoint

Run Details

1 of 15 new or added lines in 3 files covered. (6.67%)

1102 of 1415 relevant lines covered (77.88%)

0.78 hits per line

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

0.0

/trailblazer/server/api.py

import datetime
import multiprocessing
import os
from http import HTTPStatus
from typing import Mapping

from flask import Blueprint, Response, abort, g, jsonify, make_response, request
from google.auth import jwt
from http import HTTPStatus
from pydantic import ValidationError
from sqlalchemy.orm import Query

from trailblazer.constants import (
    ONE_MONTH_IN_DAYS,
    TRAILBLAZER_TIME_STAMP,
    TrailblazerStatus,
)
from trailblazer.server.ext import store
from trailblazer.server.schemas import AnalysisUpdate
from trailblazer.store.models import Analysis, Info, User
from trailblazer.utils.datetime import get_date_number_of_days_ago

ANALYSIS_HOST: str = os.environ.get("ANALYSIS_HOST")

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


def stringify_timestamps(data: dict) -> dict[str, str]:
    """Convert datetime into string before dumping in order to avoid information loss"""
    for key, val in data.items():
        if isinstance(val, datetime.datetime):
            data[key] = str(val)
    return data


@blueprint.before_request
def before_request():
    """Authentication that is run before processing requests to the application"""
    if request.method == "OPTIONS":
        return make_response(jsonify(ok=True), 204)
    if os.environ.get("SCOPE") == "DEVELOPMENT":
        return
    auth_header = request.headers.get("Authorization")
    if auth_header:
        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)
    user: User = store.get_user(email=user_data["email"], exclude_archived=True)
    if not user:
        return abort(403, f"{user_data['email']} doesn't have access")
    g.current_user = user


@blueprint.route("/analyses")
def analyses():
    """Display analyses."""
    per_page = int(request.args.get("per_page", 100))
    page = int(request.args.get("page", 1))
    analyses: Query = store.get_analyses_query_by_search_term_and_is_visible(
        search_term=request.args.get("query"),
        is_visible=bool(request.args.get("is_visible")),
    )

    query_page: Query = store.paginate_query(query=analyses, page=page, per_page=per_page)
    response_data = []
    for analysis in query_page.all():
        analysis_data = analysis.to_dict()
        analysis_data["user"] = analysis.user.to_dict() if analysis.user else None
        response_data.append(analysis_data)
    return jsonify(analyses=response_data)


@blueprint.route("/analyses/<int:analysis_id>", methods=["GET", "PUT"])
def analysis(analysis_id):
    """Retrieve or update an analysis."""
    analysis: Analysis = store.get_analysis_with_id(analysis_id)
    if analysis is None:
        return abort(404)

    if request.method == "PUT":
        try:
            analysis_update = AnalysisUpdate.model_validate(request.json)
            store.update_analysis(
                analysis_id=analysis_id,
                comment=analysis_update.comment,
                status=analysis_update.status,
                is_visible=analysis_update.is_visible,
            )
        except ValidationError as error:
            return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST

    data = analysis.to_dict()
    data["jobs"] = [job.to_dict() for job in analysis.jobs]
    data["user"] = analysis.user.to_dict() if analysis.user else None
    return jsonify(**data)


@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")
def aggregate_jobs():
    """Return stats about failed jobs."""
    time_window: datetime = get_date_number_of_days_ago(
        number_of_days_ago=int(request.args.get("days_back", ONE_MONTH_IN_DAYS))
    )
    failed_jobs: list[dict[str, str | int]] = store.get_nr_jobs_with_status_per_category(
        status=TrailblazerStatus.FAILED, since_when=time_window
    )
    return jsonify(jobs=failed_jobs)


@blueprint.route("/update-all")
def update_analyses():
    """Update all ongoing analysis by querying SLURM."""
    process = multiprocessing.Process(
        target=store.update_ongoing_analyses,
        kwargs={"analysis_host": ANALYSIS_HOST},
    )
    process.start()
    return jsonify(f"Success! Trailblazer updated {datetime.datetime.now()}"), HTTPStatus.CREATED


@blueprint.route("/update/<int:analysis_id>", methods=["PUT"])
def update_analysis(analysis_id):
    """Update a specific analysis."""
    try:
        process = multiprocessing.Process(
            target=store.update_run_status,
            kwargs={"analysis_id": analysis_id, "analysis_host": ANALYSIS_HOST},
        )
        process.start()
        return jsonify("Success! Update request sent"), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT


@blueprint.route("/cancel/<int:analysis_id>", methods=["PUT"])
def cancel(analysis_id):
    """Cancel an analysis and all slurm jobs associated with it."""
    auth_header = request.headers.get("Authorization")
    jwt_token = auth_header.split("Bearer ")[-1]
    user_data = jwt.decode(jwt_token, verify=False)
    try:
        process = multiprocessing.Process(
            target=store.cancel_ongoing_analysis,
            kwargs={
                "analysis_id": analysis_id,
                "analysis_host": ANALYSIS_HOST,
                "email": user_data["email"],
            },
        )
        process.start()
        return jsonify("Success! Cancel request sent"), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT


@blueprint.route("/delete/<int:analysis_id>", methods=["PUT"])
def delete(analysis_id):
    """Delete an analysis and all slurm jobs associated with it."""
    try:
        process = multiprocessing.Process(
            target=store.delete_analysis,
            kwargs={"analysis_id": analysis_id, "force": True},
        )
        process.start()
        return jsonify("Success! Delete request sent!"), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT


# 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: Response.json = request.json
    latest_case_analysis: Analysis | None = store.get_latest_analysis_for_case(
        case_id=post_request.get("case_id")
    )
    if latest_case_analysis:
        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("/find-analysis", methods=["POST"])
def post_find_analysis():
    """Find analysis using case id, date, and status."""
    post_request: Response.json = request.json
    analysis: Analysis = store.get_analysis(
        case_id=post_request.get("case_id"),
        started_at=datetime.strptime(post_request.get("started_at"), TRAILBLAZER_TIME_STAMP).date(),
        status=post_request.get("status"),
    )
    if analysis:
        raw_analysis: dict[str, str] = stringify_timestamps(analysis.to_dict())
        return jsonify(**raw_analysis), HTTPStatus.OK
    return jsonify(None), HTTPStatus.OK


@blueprint.route("/delete-analysis", methods=["POST"])
def post_delete_analysis():
    """Delete analysis using analysis_id. If analysis is ongoing, an error will be raised.
    To delete ongoing analysis, --force flag should also be passed.
    If an ongoing analysis is deleted in ths manner, all ongoing jobs will be cancelled"""
    post_request: Response.json = request.json
    try:
        store.delete_analysis(
            analysis_id=post_request.get("analysis_id"), force=post_request.get("force")
        )
        return jsonify(None), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT


@blueprint.route("/mark-analyses-deleted", methods=["POST"])
def post_mark_analyses_deleted():
    """Mark all analysis belonging to a case as deleted."""
    post_request: Response.json = request.json
    case_analyses: list[Analysis] | None = store.update_case_analyses_as_deleted(
        case_id=post_request.get("case_id")
    )
    raw_analysis = [
        stringify_timestamps(case_analysis.to_dict()) for case_analysis in case_analyses
    ]
    if raw_analysis:
        return jsonify(*raw_analysis), HTTPStatus.CREATED
    return jsonify(None), HTTPStatus.CREATED


@blueprint.route("/add-pending-analysis", methods=["POST"])
def post_add_pending_analysis():
    """Add new analysis with status: pending."""
    post_request: Response.json = request.json
    try:
        analysis: Analysis = store.add_pending_analysis(
            case_id=post_request.get("case_id"),
            email=post_request.get("email"),
            type=post_request.get("type"),
            config_path=post_request.get("config_path"),
            out_dir=post_request.get("out_dir"),
            priority=post_request.get("priority"),
            data_analysis=post_request.get("data_analysis"),
            ticket_id=post_request.get("ticket"),
            workflow_manager=post_request.get("workflow_manager"),
        )
        raw_analysis: dict = stringify_timestamps(analysis.to_dict())
        return jsonify(**raw_analysis), 201
    except Exception as exception:
        return jsonify(f"Exception: {exception}"), 409


@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:
        store.update_analysis_status(
            case_id=put_request.get("case_id"), status=put_request.get("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:
        store.update_analysis_comment(
            case_id=put_request.get("case_id"), comment=put_request.get("comment")
        )
        return jsonify("Success! Adding comment request sent"), HTTPStatus.CREATED
    except Exception as error:
        return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT

1	import datetime	×
2	import multiprocessing	×
3	import os	×
4	from http import HTTPStatus	×
5	from typing import Mapping	×
6
7	from flask import Blueprint, Response, abort, g, jsonify, make_response, request	×
8	from google.auth import jwt	×
NEW 9	from http import HTTPStatus	×
NEW 10	from pydantic import ValidationError	×
11	from sqlalchemy.orm import Query	×
12
13	from trailblazer.constants import (	×
14	ONE_MONTH_IN_DAYS,
15	TRAILBLAZER_TIME_STAMP,
16	TrailblazerStatus,
17	)
18	from trailblazer.server.ext import store	×
NEW 19	from trailblazer.server.schemas import AnalysisUpdate	×
20	from trailblazer.store.models import Analysis, Info, User	×
21	from trailblazer.utils.datetime import get_date_number_of_days_ago	×
22
23	ANALYSIS_HOST: str = os.environ.get("ANALYSIS_HOST")	×
24
25	blueprint = Blueprint("api", __name__, url_prefix="/api/v1")	×
26
27
28	def stringify_timestamps(data: dict) -> dict[str, str]:	×
29	"""Convert datetime into string before dumping in order to avoid information loss"""
30	for key, val in data.items():	×
31	if isinstance(val, datetime.datetime):	×
32	data[key] = str(val)	×
33	return data	×
34
35
36	@blueprint.before_request	×
37	def before_request():	×
38	"""Authentication that is run before processing requests to the application"""
39	if request.method == "OPTIONS":	×
40	return make_response(jsonify(ok=True), 204)	×
41	if os.environ.get("SCOPE") == "DEVELOPMENT":	×
42	return	×
43	auth_header = request.headers.get("Authorization")	×
44	if auth_header:	×
45	jwt_token = auth_header.split("Bearer ")[-1]	×
46	else:
47	return abort(403, "no JWT token found on request")	×
48
49	user_data: Mapping = jwt.decode(jwt_token, verify=False)	×
50	user: User = store.get_user(email=user_data["email"], exclude_archived=True)	×
51	if not user:	×
52	return abort(403, f"{user_data['email']} doesn't have access")	×
53	g.current_user = user	×
54
55
56	@blueprint.route("/analyses")	×
57	def analyses():	×
58	"""Display analyses."""
59	per_page = int(request.args.get("per_page", 100))	×
60	page = int(request.args.get("page", 1))	×
61	analyses: Query = store.get_analyses_query_by_search_term_and_is_visible(	×
62	search_term=request.args.get("query"),
63	is_visible=bool(request.args.get("is_visible")),
64	)
65
66	query_page: Query = store.paginate_query(query=analyses, page=page, per_page=per_page)	×
67	response_data = []	×
68	for analysis in query_page.all():	×
69	analysis_data = analysis.to_dict()	×
70	analysis_data["user"] = analysis.user.to_dict() if analysis.user else None	×
71	response_data.append(analysis_data)	×
72	return jsonify(analyses=response_data)	×
73
74
75	@blueprint.route("/analyses/<int:analysis_id>", methods=["GET", "PUT"])	×
76	def analysis(analysis_id):	×
77	"""Retrieve or update an analysis."""
78	analysis: Analysis = store.get_analysis_with_id(analysis_id)	×
79	if analysis is None:	×
80	return abort(404)	×
81
82	if request.method == "PUT":	×
NEW 83	try:	×
NEW 84	analysis_update = AnalysisUpdate.model_validate(request.json)	×
NEW 85	store.update_analysis(	×
86	analysis_id=analysis_id,
87	comment=analysis_update.comment,
88	status=analysis_update.status,
89	is_visible=analysis_update.is_visible,
90	)
NEW 91	except ValidationError as error:	×
NEW 92	return jsonify(error=str(error)), HTTPStatus.BAD_REQUEST	×
93
94	data = analysis.to_dict()	×
95	data["jobs"] = [job.to_dict() for job in analysis.jobs]	×
96	data["user"] = analysis.user.to_dict() if analysis.user else None	×
97	return jsonify(**data)	×
98
99
100	@blueprint.route("/info")	×
101	def info():	×
102	"""Display metadata about database."""
103	info: Info = store.get_query(table=Info).first()	×
104	return jsonify(**info.to_dict())	×
105
106
107	@blueprint.route("/me")	×
108	def me():	×
109	"""Return information about a logged in user."""
110	return jsonify(**g.current_user.to_dict())	×
111
112
113	@blueprint.route("/aggregate/jobs")	×
114	def aggregate_jobs():	×
115	"""Return stats about failed jobs."""
116	time_window: datetime = get_date_number_of_days_ago(	×
117	number_of_days_ago=int(request.args.get("days_back", ONE_MONTH_IN_DAYS))
118	)
119	failed_jobs: list[dict[str, str \| int]] = store.get_nr_jobs_with_status_per_category(	×
120	status=TrailblazerStatus.FAILED, since_when=time_window
121	)
122	return jsonify(jobs=failed_jobs)	×
123
124
125	@blueprint.route("/update-all")	×
126	def update_analyses():	×
127	"""Update all ongoing analysis by querying SLURM."""
128	process = multiprocessing.Process(	×
129	target=store.update_ongoing_analyses,
130	kwargs={"analysis_host": ANALYSIS_HOST},
131	)
132	process.start()	×
133	return jsonify(f"Success! Trailblazer updated {datetime.datetime.now()}"), HTTPStatus.CREATED	×
134
135
136	@blueprint.route("/update/<int:analysis_id>", methods=["PUT"])	×
137	def update_analysis(analysis_id):	×
138	"""Update a specific analysis."""
139	try:	×
140	process = multiprocessing.Process(	×
141	target=store.update_run_status,
142	kwargs={"analysis_id": analysis_id, "analysis_host": ANALYSIS_HOST},
143	)
144	process.start()	×
145	return jsonify("Success! Update request sent"), HTTPStatus.CREATED	×
146	except Exception as error:	×
147	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
148
149
150	@blueprint.route("/cancel/<int:analysis_id>", methods=["PUT"])	×
151	def cancel(analysis_id):	×
152	"""Cancel an analysis and all slurm jobs associated with it."""
153	auth_header = request.headers.get("Authorization")	×
154	jwt_token = auth_header.split("Bearer ")[-1]	×
155	user_data = jwt.decode(jwt_token, verify=False)	×
156	try:	×
157	process = multiprocessing.Process(	×
158	target=store.cancel_ongoing_analysis,
159	kwargs={
160	"analysis_id": analysis_id,
161	"analysis_host": ANALYSIS_HOST,
162	"email": user_data["email"],
163	},
164	)
165	process.start()	×
166	return jsonify("Success! Cancel request sent"), HTTPStatus.CREATED	×
167	except Exception as error:	×
168	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
169
170
171	@blueprint.route("/delete/<int:analysis_id>", methods=["PUT"])	×
172	def delete(analysis_id):	×
173	"""Delete an analysis and all slurm jobs associated with it."""
174	try:	×
175	process = multiprocessing.Process(	×
176	target=store.delete_analysis,
177	kwargs={"analysis_id": analysis_id, "force": True},
178	)
179	process.start()	×
180	return jsonify("Success! Delete request sent!"), HTTPStatus.CREATED	×
181	except Exception as error:	×
182	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
183
184
185	# CG REST INTERFACE ###
186	# ONLY POST routes which accept messages in specific format
187	# NOT for use with GUI (for now)
188
189
190	@blueprint.route("/get-latest-analysis", methods=["POST"])	×
191	def post_get_latest_analysis():	×
192	"""Return latest analysis entry for specified case id."""
193	post_request: Response.json = request.json	×
194	latest_case_analysis: Analysis \| None = store.get_latest_analysis_for_case(	×
195	case_id=post_request.get("case_id")
196	)
197	if latest_case_analysis:	×
198	raw_analysis: dict[str, str] = stringify_timestamps(latest_case_analysis.to_dict())	×
199	return jsonify(**raw_analysis), HTTPStatus.OK	×
200	return jsonify(None), HTTPStatus.OK	×
201
202
203	@blueprint.route("/find-analysis", methods=["POST"])	×
204	def post_find_analysis():	×
205	"""Find analysis using case id, date, and status."""
206	post_request: Response.json = request.json	×
207	analysis: Analysis = store.get_analysis(	×
208	case_id=post_request.get("case_id"),
209	started_at=datetime.strptime(post_request.get("started_at"), TRAILBLAZER_TIME_STAMP).date(),
210	status=post_request.get("status"),
211	)
212	if analysis:	×
213	raw_analysis: dict[str, str] = stringify_timestamps(analysis.to_dict())	×
214	return jsonify(**raw_analysis), HTTPStatus.OK	×
215	return jsonify(None), HTTPStatus.OK	×
216
217
218	@blueprint.route("/delete-analysis", methods=["POST"])	×
219	def post_delete_analysis():	×
220	"""Delete analysis using analysis_id. If analysis is ongoing, an error will be raised.
221	To delete ongoing analysis, --force flag should also be passed.
222	If an ongoing analysis is deleted in ths manner, all ongoing jobs will be cancelled"""
223	post_request: Response.json = request.json	×
224	try:	×
225	store.delete_analysis(	×
226	analysis_id=post_request.get("analysis_id"), force=post_request.get("force")
227	)
228	return jsonify(None), HTTPStatus.CREATED	×
229	except Exception as error:	×
230	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
231
232
233	@blueprint.route("/mark-analyses-deleted", methods=["POST"])	×
234	def post_mark_analyses_deleted():	×
235	"""Mark all analysis belonging to a case as deleted."""
236	post_request: Response.json = request.json	×
237	case_analyses: list[Analysis] \| None = store.update_case_analyses_as_deleted(	×
238	case_id=post_request.get("case_id")
239	)
240	raw_analysis = [	×
241	stringify_timestamps(case_analysis.to_dict()) for case_analysis in case_analyses
242	]
243	if raw_analysis:	×
244	return jsonify(*raw_analysis), HTTPStatus.CREATED	×
245	return jsonify(None), HTTPStatus.CREATED	×
246
247
248	@blueprint.route("/add-pending-analysis", methods=["POST"])	×
249	def post_add_pending_analysis():	×
250	"""Add new analysis with status: pending."""
251	post_request: Response.json = request.json	×
252	try:	×
253	analysis: Analysis = store.add_pending_analysis(	×
254	case_id=post_request.get("case_id"),
255	email=post_request.get("email"),
256	type=post_request.get("type"),
257	config_path=post_request.get("config_path"),
258	out_dir=post_request.get("out_dir"),
259	priority=post_request.get("priority"),
260	data_analysis=post_request.get("data_analysis"),
261	ticket_id=post_request.get("ticket"),
262	workflow_manager=post_request.get("workflow_manager"),
263	)
264	raw_analysis: dict = stringify_timestamps(analysis.to_dict())	×
265	return jsonify(**raw_analysis), 201	×
266	except Exception as exception:	×
267	return jsonify(f"Exception: {exception}"), 409	×
268
269
270	@blueprint.route("/set-analysis-uploaded", methods=["PUT"])	×
271	def set_analysis_uploaded():	×
272	"""Set the analysis uploaded at attribute."""
273	put_request: Response.json = request.json	×
274	try:	×
275	store.update_analysis_uploaded_at(	×
276	case_id=put_request.get("case_id"), uploaded_at=put_request.get("uploaded_at")
277	)
278	return jsonify("Success! Uploaded at request sent"), HTTPStatus.CREATED	×
279	except Exception as error:	×
280	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
281
282
283	@blueprint.route("/set-analysis-status", methods=["PUT"])	×
284	def set_analysis_status():	×
285	"""Update analysis status of a case with supplied status."""
286	put_request: Response.json = request.json	×
287	try:	×
288	store.update_analysis_status(	×
289	case_id=put_request.get("case_id"), status=put_request.get("status")
290	)
291	return (	×
292	jsonify(f"Success! Analysis set to {put_request.get('status')} request sent"),
293	HTTPStatus.CREATED,
294	)
295	except Exception as error:	×
296	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×
297
298
299	@blueprint.route("/add-comment", methods=["PUT"])	×
300	def add_comment():	×
301	"""Updating comment on analysis."""
302	put_request: Response.json = request.json	×
303	try:	×
304	store.update_analysis_comment(	×
305	case_id=put_request.get("case_id"), comment=put_request.get("comment")
306	)
307	return jsonify("Success! Adding comment request sent"), HTTPStatus.CREATED	×
308	except Exception as error:	×
309	return jsonify(f"Exception: {error}"), HTTPStatus.CONFLICT	×

Clinical-Genomics / trailblazer / 7005273242

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