Fix UniProt TrEMBL parsing, fetcher error chaining and id_mapping

.github/workflows/lint.yml

@@ -8,6 +8,8 @@ jobs:
       - uses: chartboost/ruff-action@v1
         with:
           src: './pyenzyme'
+          version: 0.15.22  # keep in sync with .pre-commit-config.yaml
       - uses: chartboost/ruff-action@v1
         with:
           src: './tests'
+          version: 0.15.22

pyenzyme/composer.py

@@ -1,4 +1,3 @@
-import re
 from typing import List, Optional, Tuple, Callable, Any
 
 from rich.console import Console
@@ -98,21 +97,20 @@ def compose(
     reaction_objects.sort(key=lambda x: x.id)
 
     if id_mapping:
-        small_molecule_objects = _apply_id_mapping(
-            small_molecule_objects,
-            id_mapping,
-            r"CHEBI|PUBCHEM",
-        )
-        protein_objects = _apply_id_mapping(
-            protein_objects,
-            id_mapping,
-            r"UNIPROT|PDB",
-        )
-        reaction_objects = _apply_id_mapping(
-            reaction_objects,
-            id_mapping,
-            r"RHEA",
-        )
+        renamed = _apply_id_mapping(small_molecule_objects + protein_objects, id_mapping)
+        _apply_id_mapping(reaction_objects, id_mapping)
+
+        species_ids = [obj.id for obj in small_molecule_objects + protein_objects]
+        duplicates = sorted({i for i in species_ids if species_ids.count(i) > 1})
+        if duplicates:
+            raise ValueError(
+                f"id_mapping gives several species the same ID: {', '.join(duplicates)}"
+            )
+
+        # Keep reactions pointing at the renamed species
+        for reaction in reaction_objects:
+            for element in reaction.reactants + reaction.products + reaction.modifiers:
+                element.species_id = renamed.get(element.species_id, element.species_id)
 
     # Set vessel IDs if vessel is provided
     vessels = []
@@ -198,18 +196,18 @@ def _fetch_with_fetchers(
         ValueError: If no fetcher can handle the given entity ID
     """
     errors = []
+    last_error = None
     for fetcher in fetchers:
         try:
             return fetcher(entity_id)
         except Exception as e:
             errors.append(f"{fetcher.__name__}: {type(e).__name__}: {e}")
-            continue
+            last_error = e
 
     detail = "; ".join(errors)
     raise ValueError(
-        f"No {entity_type} fetcher succeeded for {entity_id}. "
-        f"Tried: {detail}"
-    )
+        f"No {entity_type} fetcher succeeded for {entity_id}. Tried: {detail}"
+    ) from last_error
 
 
 def _remove_duplicates(objects: List[Any]) -> List[Any]:
@@ -239,29 +237,32 @@ def _remove_duplicates(objects: List[Any]) -> List[Any]:
 def _apply_id_mapping(
     objects: List[Any],
     id_mapping: dict[str, str],
-    prefix: str,
-) -> List[Any]:
+) -> dict[str, str]:
     """
-    Apply ID mapping to objects based on the provided mapping.
+    Apply ID mapping to objects (in place) based on the provided mapping.
+
+    A key matches an object when it equals the object's ``ld_id`` or a trailing
+    part of it, ignoring case: "CHEBI:63153", "chebi:63153" and "63153" all
+    match "OBO:CHEBI_63153", but "CHEBI:1" does not match "OBO:CHEBI_15377".
 
     Args:
         objects: List of objects to apply ID mapping to
         id_mapping: Dictionary mapping old IDs to new IDs
-        prefix: Regex pattern to match ID prefixes
 
     Returns:
-        List of objects with updated IDs
+        Dictionary of the renames applied, from previous to new object ID
     """
+    renamed = {}
     for obj in objects:
-        key = next(
-            (
-                k
-                for k in id_mapping
-                if re.sub(prefix + ":", "", k) in obj.ld_id.replace("_", ":")
-            ),
-            None,
-        )
+        ld_id = _normalize_id(obj.ld_id)
+        key = next((k for k in id_mapping if ld_id.endswith(_normalize_id(k))), None)
         if key:
+            renamed[obj.id] = id_mapping[key]
             obj.id = id_mapping[key]
 
-    return objects
+    return renamed
+
+
+def _normalize_id(identifier: str) -> str:
+    """'OBO:CHEBI_63153' -> ':OBO:CHEBI:63153', 'pubchem:CID962' -> ':PUBCHEM:962'"""
+    return (":" + identifier.upper().replace("_", ":")).replace(":CID", ":")

pyenzyme/fetcher/uniprot.py

@@ -7,7 +7,7 @@
 
 import requests
 from typing import List, Optional, Union
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 from pyenzyme.fetcher.chebi import process_id
 from pyenzyme.versions import v2
 
@@ -25,31 +25,43 @@ class ProteinName(BaseModel):
 
 
 class RecommendedName(BaseModel):
-    """Model for recommended protein name in UniProt API"""
+    """Model for a recommended or submitted protein name in UniProt API"""
 
     full_name: ProteinName = Field(alias="fullName")
-    ec_numbers: Optional[List[ECNumber]] = Field(alias="ecNumbers")
+    ec_numbers: Optional[List[ECNumber]] = Field(default=None, alias="ecNumbers")
 
 
 class ProteinDescription(BaseModel):
     """Model for protein description in UniProt API"""
 
-    recommended_name: Optional[RecommendedName] = Field(alias="recommendedName")
+    recommended_name: Optional[RecommendedName] = Field(
+        default=None, alias="recommendedName"
+    )
+    submission_names: List[RecommendedName] = Field(
+        default_factory=list, alias="submissionNames"
+    )
+
+    @property
+    def name(self) -> Optional[RecommendedName]:
+        """Recommended name, or the first submission name for unreviewed (TrEMBL) entries."""
+        if self.recommended_name:
+            return self.recommended_name
+        return next(iter(self.submission_names), None)
 
 
 class Organism(BaseModel):
     """Model for organism in UniProt API"""
 
-    scientific_name: Optional[str] = Field(alias="scientificName")
-    taxon_id: Optional[int] = Field(alias="taxonId")
+    scientific_name: Optional[str] = Field(default=None, alias="scientificName")
+    taxon_id: Optional[int] = Field(default=None, alias="taxonId")
 
 
 class Sequence(BaseModel):
     """Model for protein sequence in UniProt API"""
 
     value: str
     length: int
-    mol_weight: Optional[int] = Field(alias="molWeight")
+    mol_weight: Optional[int] = Field(default=None, alias="molWeight")
 
 
 class UniProtEntry(BaseModel):
@@ -62,7 +74,7 @@ class UniProtEntry(BaseModel):
     organism: Optional[Organism] = None
     sequence: Optional[Sequence] = None
     accession: str = Field(alias="primaryAccession")
-    annotation_score: Optional[float] = Field(alias="annotationScore")
+    annotation_score: Optional[float] = Field(default=None, alias="annotationScore")
 
 
 class UniProtClient:
@@ -88,26 +100,37 @@ def get_entry_by_id(self, uniprot_id: str) -> Union[UniProtEntry, None]:
             ValueError: If the UniProt ID is invalid or not found
             ConnectionError: If the connection to the UniProt server fails
         """
-        # Construct the URL
         url = f"{self.BASE_URL}/{uniprot_id}.json"
 
         try:
             response = requests.get(url)
             response.raise_for_status()
-
-            if response.status_code == 200:
-                try:
-                    data = response.json()
-                    return UniProtEntry.model_validate(data)
-                except Exception as e:
-                    raise ValueError(f"Failed to parse UniProt response: {str(e)}")
-            else:
+        except requests.exceptions.HTTPError as e:
+            # 4xx means a bad or unknown ID, not a connection problem
+            if e.response is not None and e.response.status_code < 500:
                 raise ValueError(
-                    f"Failed to retrieve UniProt entry for ID {uniprot_id}"
-                )
-
+                    f"UniProt ID '{uniprot_id}' is invalid or not found "
+                    f"(HTTP {e.response.status_code})"
+                ) from e
+            raise ConnectionError(f"UniProt server error: {e}") from e
         except requests.exceptions.RequestException as e:
-            raise ConnectionError(f"Connection to UniProt server failed: {str(e)}")
+            raise ConnectionError(f"Connection to UniProt server failed: {e}") from e
+
+        data = response.json()
+
+        if data.get("entryType") == "Inactive":
+            reason = data.get("inactiveReason", {})
+            detail = reason.get("inactiveReasonType", "unknown reason")
+            if "deletedReason" in reason:
+                detail += f": {reason['deletedReason']}"
+            if reason.get("mergeDemergeTo"):
+                detail += f", replaced by {', '.join(reason['mergeDemergeTo'])}"
+            raise ValueError(f"UniProt entry {uniprot_id} is inactive ({detail})")
+
+        try:
+            return UniProtEntry.model_validate(data)
+        except ValidationError as e:
+            raise ValueError(f"Failed to parse UniProt response: {e}") from e
 
 
 def fetch_uniprot(
@@ -131,6 +154,7 @@ def fetch_uniprot(
     client = UniProtClient()
 
     # Allow prefixing with 'uniprot:'
+    uniprot_id = uniprot_id.strip()
     if uniprot_id.lower().startswith("uniprot:"):
         uniprot_id = uniprot_id.split(":", 1)[-1]
 
@@ -139,14 +163,9 @@ def fetch_uniprot(
     if not uniprot_entry:
         raise ValueError(f"No data found for UniProt ID {uniprot_id}")
 
-    # Extract protein name
-    name = uniprot_id
-    if (
-        uniprot_entry.protein_description
-        and uniprot_entry.protein_description.recommended_name
-        and uniprot_entry.protein_description.recommended_name.full_name
-    ):
-        name = uniprot_entry.protein_description.recommended_name.full_name.value
+    protein_name = uniprot_entry.protein_description.name
+
+    name = protein_name.full_name.value if protein_name else uniprot_id
 
     # Extract sequence if available
     sequence = None
@@ -163,26 +182,13 @@ def fetch_uniprot(
 
     # Extract EC number
     ecnumber = None
-    if (
-        uniprot_entry.protein_description
-        and uniprot_entry.protein_description.recommended_name
-        and uniprot_entry.protein_description.recommended_name.ec_numbers
-        and len(uniprot_entry.protein_description.recommended_name.ec_numbers) > 0
-    ):
-        ecnumber = uniprot_entry.protein_description.recommended_name.ec_numbers[
-            0
-        ].value
+    if protein_name and protein_name.ec_numbers:
+        ecnumber = protein_name.ec_numbers[0].value
 
     # Create a Protein instance
     if protein_id is None:
-        if (
-            uniprot_entry.protein_description
-            and uniprot_entry.protein_description.recommended_name
-            and uniprot_entry.protein_description.recommended_name.full_name
-        ):
-            protein_id = process_id(
-                uniprot_entry.protein_description.recommended_name.full_name.value
-            )
+        if protein_name:
+            protein_id = process_id(protein_name.full_name.value)
         else:
             protein_id = uniprot_entry.accession
 

tests/integration/test_composer.py

@@ -58,6 +58,19 @@ def test_compose_no_vessel(self):
         )
         assert to_dict_wo_json_ld(doc) == to_dict_wo_json_ld(expected_doc)
 
+    def test_compose_id_mapping_renames_reaction_species(self):
+        doc = pe.compose(
+            name="test",
+            reactions=["RHEA:19273"],
+            id_mapping={"CHEBI:63153": "ManNAc"},
+        )
+
+        declared = {sm.id for sm in doc.small_molecules}
+        assert "ManNAc" in declared
+        for reaction in doc.reactions:
+            for element in reaction.reactants + reaction.products:
+                assert element.species_id in declared
+
     def test_compose_invalid_id(self):
         # Act
         # Invalid Protein ID

tests/integration/test_fetcher.py

@@ -97,6 +97,15 @@ def test_fetch_uniprot_to_protein_with_id(self):
         assert len(protein.references) == 1
         assert protein.references[0] == "https://www.uniprot.org/uniprotkb/P07327"
 
+    @pytest.mark.remote
+    def test_fetch_uniprot_unreviewed_entry(self):
+        # TrEMBL entry: has submissionNames but no recommendedName
+        protein = fetch_uniprot("Q7DDU0")
+
+        assert protein.name == "Polysialic acid capsule biosynthesis protein SiaC"
+        assert protein.ld_id == "uniprot:Q7DDU0"
+        assert protein.sequence
+
     @pytest.mark.remote
     def test_fetch_rhea_to_reaction(self):
         reaction, small_molecules = fetch_rhea("RHEA:22864")