From 589c0504a08c66e8c4d2211a03193c7304460105 Mon Sep 17 00:00:00 2001 From: Sally Grindstaff Date: Mon, 26 Aug 2024 09:58:18 -0700 Subject: [PATCH] Add error handling to api map function --- src/api/routers/map.py | 129 +++++++++++++++++++++++++++++------------ 1 file changed, 91 insertions(+), 38 deletions(-) diff --git a/src/api/routers/map.py b/src/api/routers/map.py index ca354d4..3f2ec66 100644 --- a/src/api/routers/map.py +++ b/src/api/routers/map.py @@ -1,21 +1,26 @@ +""""Provide mapping router""" from cool_seq_tool.schemas import AnnotationLayer -from fastapi import APIRouter +from fastapi import APIRouter, HTTPException +from requests import HTTPError -from dcd_mapping.align import align +from dcd_mapping.align import AlignmentError, BlatNotFoundError, align from dcd_mapping.annotate import ( _get_computed_reference_sequence, _get_mapped_reference_sequence, _set_scoreset_layer, annotate, ) +from dcd_mapping.lookup import DataLookupError from dcd_mapping.mavedb_data import ( + ScoresetNotSupportedError, get_raw_scoreset_metadata, get_scoreset_metadata, get_scoreset_records, ) -from dcd_mapping.schemas import ScoreAnnotation, ScoresetMapping -from dcd_mapping.transcripts import select_transcript -from dcd_mapping.vrs_map import vrs_map +from dcd_mapping.resource_utils import ResourceAcquisitionError +from dcd_mapping.schemas import ScoreAnnotation, ScoresetMapping, VrsVersion +from dcd_mapping.transcripts import TxSelectError, select_transcript +from dcd_mapping.vrs_map import VrsMapError, vrs_map router = APIRouter( prefix="/api/v1", tags=["mappings"], responses={404: {"description": "Not found"}} @@ -24,48 +29,98 @@ @router.post(path="/map/{urn}", status_code=200, response_model=ScoresetMapping) async def map_scoreset(urn: str) -> ScoresetMapping: - metadata = get_scoreset_metadata(urn) - records = get_scoreset_records(urn, True) + """Perform end-to-end mapping for a scoreset. - alignment_result = align(metadata, True) + :param urn: identifier for a scoreset. + :param output_path: optional path to save output at + :param vrs_version: version of VRS objects to output (1.3 or 2) + :param silent: if True, suppress console information output + """ + try: + metadata = get_scoreset_metadata(urn) + records = get_scoreset_records(urn, True) + except ScoresetNotSupportedError as e: + return ScoresetMapping( + metadata=None, + error_message=str(e).strip("'"), + ) + except ResourceAcquisitionError as e: + msg = f"Unable to acquire resource from MaveDB: {e}" + raise HTTPException(status_code=500, detail=msg) from e - transcript = await select_transcript(metadata, records, alignment_result) + try: + alignment_result = align(metadata, True) + except BlatNotFoundError as e: + msg = "BLAT command appears missing. Ensure it is available on the $PATH or use the environment variable BLAT_BIN_PATH to point to it. See instructions in the README prerequisites section for more." + raise HTTPException(status_code=500, detail=msg) from e + except ResourceAcquisitionError as e: + msg = f"BLAT resource could not be acquired: {e}" + raise HTTPException(status_code=500, detail=msg) from e + except AlignmentError as e: + return ScoresetMapping(metadata=metadata, error_message=str(e).strip("'")) - vrs_results = vrs_map(metadata, alignment_result, records, transcript, True) + try: + transcript = await select_transcript(metadata, records, alignment_result) + except (TxSelectError, KeyError, ValueError) as e: + return ScoresetMapping(metadata=metadata, error_message=str(e).strip("'")) + except HTTPError as e: + msg = f"HTTP error occurred during transcript selection: {e}" + raise HTTPException(status_code=500, detail=msg) from e + except DataLookupError as e: + msg = f"Data lookup error occurred during transcript selection: {e}" + raise HTTPException(status_code=500, detail=msg) from e - # TODO raise server error if vrs_results is None + try: + vrs_results = vrs_map(metadata, alignment_result, records, transcript, True) + except VrsMapError as e: + return ScoresetMapping(metadata=metadata, error_message=str(e).strip("'")) if vrs_results is None: - return None + return ScoresetMapping( + metadata=metadata, + error_message="No variant mappings available for this score set", + ) - vrs_results = annotate(vrs_results, transcript, metadata) + try: + vrs_results = annotate(vrs_results, transcript, metadata, VrsVersion.V_2) + except Exception as e: + return ScoresetMapping(metadata=metadata, error_message=str(e).strip("'")) + if vrs_results is None: + return ScoresetMapping( + metadata=metadata, + error_message="No annotated variant mappings available for this score set", + ) - raw_metadata = get_raw_scoreset_metadata(urn) - # TODO change vrs map back to always use only the preferred layer - # preferred_layers = {mapping.annotation_layer for mapping in vrs_results} - preferred_layers = { - _set_scoreset_layer(urn, vrs_results), - } + try: + raw_metadata = get_raw_scoreset_metadata(urn) + preferred_layers = { + _set_scoreset_layer(urn, vrs_results), + } - reference_sequences = { - layer: {"computed_reference_sequence": None, "mapped_reference_sequence": None} - for layer in AnnotationLayer - } + reference_sequences = { + layer: { + "computed_reference_sequence": None, + "mapped_reference_sequence": None, + } + for layer in AnnotationLayer + } - for layer in preferred_layers: - reference_sequences[layer][ - "computed_reference_sequence" - ] = _get_computed_reference_sequence(urn, layer, transcript) - reference_sequences[layer][ - "mapped_reference_sequence" - ] = _get_mapped_reference_sequence(layer, transcript, alignment_result) + for layer in preferred_layers: + reference_sequences[layer][ + "computed_reference_sequence" + ] = _get_computed_reference_sequence(urn, layer, transcript) + reference_sequences[layer][ + "mapped_reference_sequence" + ] = _get_mapped_reference_sequence(layer, transcript, alignment_result) - mapped_scores: list[ScoreAnnotation] = [] - for m in vrs_results: - if m.annotation_layer in preferred_layers: - # drop annotation layer from mapping object - mapped_scores.append(ScoreAnnotation(**m.model_dump())) + mapped_scores: list[ScoreAnnotation] = [] + for m in vrs_results: + if m.annotation_layer in preferred_layers: + # drop annotation layer from mapping object + mapped_scores.append(ScoreAnnotation(**m.model_dump())) + except Exception as e: + return ScoresetMapping(metadata=metadata, error_message=str(e).strip("'")) - output = ScoresetMapping( + return ScoresetMapping( metadata=raw_metadata, computed_protein_reference_sequence=reference_sequences[ AnnotationLayer.PROTEIN @@ -81,5 +136,3 @@ async def map_scoreset(urn: str) -> ScoresetMapping: ], mapped_scores=mapped_scores, ) - - return output