feat: add custom exceptions for VRS mapping errors and update error handling in mapping functions

Author: bencap

src/api/routers/map.py

@@ -13,6 +13,12 @@
     _set_scoreset_layer,
     annotate,
 )
+from dcd_mapping.exceptions import (
+    MissingSequenceIdError,
+    UnsupportedReferenceSequenceNameSpaceError,
+    UnsupportedReferenceSequencePrefixError,
+    VrsMapError,
+)
 from dcd_mapping.lookup import DataLookupError
 from dcd_mapping.mavedb_data import (
     ScoresetNotSupportedError,
@@ -31,7 +37,7 @@
     VrsVersion,
 )
 from dcd_mapping.transcripts import select_transcripts
-from dcd_mapping.vrs_map import VrsMapError, vrs_map
+from dcd_mapping.vrs_map import vrs_map
 
 router = APIRouter(
     prefix="/api/v1", tags=["mappings"], responses={404: {"description": "Not found"}}
@@ -115,7 +121,12 @@ async def map_scoreset(urn: str, store_path: Path | None = None) -> JSONResponse
                 transcript=transcripts[target_gene],
                 silent=True,
             )
-    except VrsMapError as e:
+    except (
+        UnsupportedReferenceSequenceNameSpaceError,
+        VrsMapError,
+        UnsupportedReferenceSequencePrefixError,
+        MissingSequenceIdError,
+    ) as e:
         return JSONResponse(
             content=ScoresetMapping(
                 metadata=metadata, error_message=str(e).strip("'")

src/dcd_mapping/cli.py

@@ -7,11 +7,11 @@
 import click
 
 from dcd_mapping.align import AlignmentError
+from dcd_mapping.exceptions import VrsMapError
 from dcd_mapping.main import map_scoreset_urn
 from dcd_mapping.resource_utils import ResourceAcquisitionError
 from dcd_mapping.schemas import VrsVersion
 from dcd_mapping.transcripts import TxSelectError
-from dcd_mapping.vrs_map import VrsMapError
 
 _logger = logging.getLogger(__name__)
 

src/dcd_mapping/exceptions.py

@@ -0,0 +1,17 @@
+"""Exceptions for DCD Mapping Module"""
+
+
+class VrsMapError(Exception):
+    """Raise in case of generic VRS mapping errors."""
+
+
+class UnsupportedReferenceSequenceNameSpaceError(ValueError):
+    """Raised when a reference sequence name space is not supported."""
+
+
+class MissingSequenceIdError(ValueError):
+    """Raised when a sequence ID is not provided."""
+
+
+class UnsupportedReferenceSequencePrefixError(ValueError):
+    """Raised when a reference sequence prefix is not supported."""

src/dcd_mapping/main.py

@@ -14,6 +14,12 @@
     save_mapped_output_json,
     write_scoreset_mapping_to_json,
 )
+from dcd_mapping.exceptions import (
+    MissingSequenceIdError,
+    UnsupportedReferenceSequenceNameSpaceError,
+    UnsupportedReferenceSequencePrefixError,
+    VrsMapError,
+)
 from dcd_mapping.lookup import (
     DataLookupError,
     check_gene_normalizer,
@@ -35,7 +41,7 @@
     VrsVersion,
 )
 from dcd_mapping.transcripts import select_transcripts
-from dcd_mapping.vrs_map import VrsMapError, vrs_map
+from dcd_mapping.vrs_map import vrs_map
 
 _logger = logging.getLogger(__name__)
 
@@ -223,7 +229,12 @@ async def map_scoreset(
                 transcript=transcripts[target_gene],
                 silent=silent,
             )
-    except VrsMapError as e:
+    except (
+        MissingSequenceIdError,
+        UnsupportedReferenceSequencePrefixError,
+        UnsupportedReferenceSequenceNameSpaceError,
+        VrsMapError,
+    ) as e:
         _emit_info(
             f"VRS mapping failed for scoreset {metadata.urn}", silent, logging.ERROR
         )

src/dcd_mapping/vrs_map.py

@@ -20,6 +20,11 @@
 from mavehgvs.util import parse_variant_strings
 from mavehgvs.variant import Variant
 
+from dcd_mapping.exceptions import (
+    MissingSequenceIdError,
+    UnsupportedReferenceSequenceNameSpaceError,
+    UnsupportedReferenceSequencePrefixError,
+)
 from dcd_mapping.lookup import (
     cdot_rest,
     get_chromosome_identifier,
@@ -37,16 +42,12 @@
 )
 from dcd_mapping.transcripts import TxSelectError
 
-__all__ = ["vrs_map", "VrsMapError"]
+__all__ = ["vrs_map"]
 
 
 _logger = logging.getLogger(__name__)
 
 
-class VrsMapError(Exception):
-    """Raise in case of VRS mapping errors."""
-
-
 def _hgvs_variant_is_valid(hgvs_string: str) -> bool:
     return not hgvs_string.endswith((".=", ")", "X"))
 
@@ -612,11 +613,8 @@ def _map_genomic(
                 error_message=str(e),
             )
     else:
-        return MappedScore(
-            accession_id=row.accession,
-            score=row.score,
-            error_message=f"Reference sequence namespace not supported: {namespace}",
-        )
+        msg = f"Unsupported reference sequence namespace: {namespace}"
+        raise UnsupportedReferenceSequenceNameSpaceError(msg)
 
     return MappedScore(
         accession_id=row.accession,
@@ -789,14 +787,8 @@ def _map_accession(
     variations: list[MappedScore] = []
     sequence_id = metadata.target_accession_id
     if sequence_id is None:
-        return [
-            MappedScore(
-                accession_id=row.accession,
-                score=row.score,
-                error_message="Could not generate mapped allele objects. No sequence id was provided.",
-            )
-            for row in records
-        ]
+        msg = " No target_accession_id was provided by target gene metadata. Target gene metadata must have a target_accession_id to map to VRS."
+        raise MissingSequenceIdError(msg)
 
     store_accession(sequence_id)
 
@@ -815,14 +807,8 @@ def _map_accession(
             hgvs_nt_mappings = _map_genomic(row, sequence_id, align_result)
             variations.append(hgvs_nt_mappings)
     else:
-        [
-            MappedScore(
-                accession_id=row.accession,
-                score=row.score,
-                error_message=f"Unrecognized accession prefix for accession id {metadata.target_accession_id}",
-            )
-            for row in records
-        ]
+        msg = f"Unrecognized accession prefix for accession id: {metadata.target_accession_id}"
+        raise UnsupportedReferenceSequencePrefixError(msg)
 
     return variations