diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index a8d77a0d..6ec1cfe8 100644 --- a/.github/workflows/lint.yml +++ b/.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 diff --git a/pyenzyme/composer.py b/pyenzyme/composer.py index e3b45e10..21db6789 100644 --- a/pyenzyme/composer.py +++ b/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", ":") diff --git a/pyenzyme/fetcher/uniprot.py b/pyenzyme/fetcher/uniprot.py index a2f019a5..fe76b7fb 100644 --- a/pyenzyme/fetcher/uniprot.py +++ b/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,23 +25,35 @@ 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): @@ -49,7 +61,7 @@ class Sequence(BaseModel): 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 diff --git a/tests/integration/test_composer.py b/tests/integration/test_composer.py index 17704596..96479f16 100644 --- a/tests/integration/test_composer.py +++ b/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 diff --git a/tests/integration/test_fetcher.py b/tests/integration/test_fetcher.py index a34fed0a..526468c1 100644 --- a/tests/integration/test_fetcher.py +++ b/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") diff --git a/tests/unit/test_composer.py b/tests/unit/test_composer.py index 61cc0e51..9461da13 100644 --- a/tests/unit/test_composer.py +++ b/tests/unit/test_composer.py @@ -1,6 +1,8 @@ import pytest -from pyenzyme.composer import _fetch_with_fetchers +import pyenzyme.composer as composer +from pyenzyme.composer import _fetch_with_fetchers, compose +from pyenzyme.versions import v2 def test_returns_first_successful_fetcher(): @@ -28,3 +30,67 @@ def not_found(_): assert "No protein fetcher succeeded for X1" in msg assert "bad_request: ValueError: 400 Bad Request" in msg assert "not_found: KeyError" in msg + + +def test_error_chains_last_fetcher_exception(): + def bad_request(_): + raise ValueError("400 Bad Request") + + def not_found(_): + raise KeyError("404") + + with pytest.raises(ValueError) as exc: + _fetch_with_fetchers("X1", [bad_request, not_found], "protein") + + assert isinstance(exc.value.__cause__, KeyError) + + +def fake_rhea(_): + mannac = v2.SmallMolecule(id="mannac", name="ManNAc", constant=False) + mannac.ld_id = "OBO:CHEBI_63153" + water = v2.SmallMolecule(id="water", name="Water", constant=False) + water.ld_id = "OBO:CHEBI_15377" + reaction = v2.Reaction(id="RHEA:19273", name="RHEA:19273", reversible=False) + reaction.add_to_reactants(species_id="mannac", stoichiometry=1) + reaction.add_to_reactants(species_id="water", stoichiometry=1) + reaction.ld_id = "rhea:19273" + return reaction, [mannac, water] + + +def compose_rhea(monkeypatch, id_mapping): + monkeypatch.setattr(composer, "REACTION_FETCHERS", [fake_rhea]) + return compose(name="test", reactions=["RHEA:19273"], id_mapping=id_mapping) + + +def test_id_mapping_renames_reaction_species(monkeypatch): + doc = compose_rhea(monkeypatch, {"CHEBI:63153": "ManNAc"}) + + assert {sm.id for sm in doc.small_molecules} == {"ManNAc", "water"} + (reaction,) = doc.reactions + assert [r.species_id for r in reaction.reactants] == ["ManNAc", "water"] + + +@pytest.mark.parametrize("key", ["chebi:63153", "63153", "OBO:CHEBI_63153"]) +def test_id_mapping_key_forms(monkeypatch, key): + doc = compose_rhea(monkeypatch, {key: "ManNAc"}) + + assert {sm.id for sm in doc.small_molecules} == {"ManNAc", "water"} + + +def test_id_mapping_does_not_match_partial_ids(monkeypatch): + # CHEBI:1 is a prefix of CHEBI:15377 (water) but a different molecule + doc = compose_rhea(monkeypatch, {"CHEBI:1": "X"}) + + assert {sm.id for sm in doc.small_molecules} == {"mannac", "water"} + + +@pytest.mark.parametrize( + "id_mapping", + [ + {"CHEBI:63153": "S", "CHEBI:15377": "S"}, # two keys, same new ID + {"CHEBI:63153": "water"}, # new ID taken by an unmapped species + ], +) +def test_id_mapping_rejects_duplicate_ids(monkeypatch, id_mapping): + with pytest.raises(ValueError, match="same ID"): + compose_rhea(monkeypatch, id_mapping) diff --git a/tests/unit/test_uniprot.py b/tests/unit/test_uniprot.py new file mode 100644 index 00000000..54a0685f --- /dev/null +++ b/tests/unit/test_uniprot.py @@ -0,0 +1,95 @@ +import pytest +import requests + +import pyenzyme.fetcher.uniprot as uniprot +from pyenzyme.fetcher.uniprot import fetch_uniprot + +# Trimmed UniProt REST response for an unreviewed (TrEMBL) entry: no recommendedName +TREMBL_ENTRY = { + "entryType": "UniProtKB unreviewed (TrEMBL)", + "primaryAccession": "Q7DDU0", + "uniProtkbId": "Q7DDU0_NEIMB", + "annotationScore": 1.0, + "organism": { + "scientificName": "Neisseria meningitidis serogroup B (strain ATCC BAA-335 / MC58)", + "taxonId": 122586, + }, + "proteinDescription": { + "submissionNames": [ + { + "fullName": {"value": "Polysialic acid capsule biosynthesis protein SiaC"}, + "ecNumbers": [{"value": "2.5.1.56"}], + } + ] + }, + "sequence": {"value": "MQNNNEF", "length": 7, "molWeight": 800}, +} + + +class FakeResponse: + def __init__(self, data=TREMBL_ENTRY, status_code=200): + self.data = data + self.status_code = status_code + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.exceptions.HTTPError(response=self) + + def json(self): + return self.data + + +def test_fetch_uniprot_falls_back_to_submission_name(monkeypatch): + monkeypatch.setattr(uniprot.requests, "get", lambda url: FakeResponse()) + + protein = fetch_uniprot("Q7DDU0") + + assert protein.name == "Polysialic acid capsule biosynthesis protein SiaC" + assert protein.id == "polysialic_acid_capsule_biosynthesis_protein_siac" + assert protein.ecnumber == "2.5.1.56" + assert protein.organism_tax_id == "122586" + assert protein.ld_id == "uniprot:Q7DDU0" + + +def test_fetch_uniprot_unknown_id_is_value_error(monkeypatch): + monkeypatch.setattr( + uniprot.requests, "get", lambda url: FakeResponse(status_code=404) + ) + + with pytest.raises(ValueError, match="invalid or not found"): + fetch_uniprot("P07327-2") + + +def test_fetch_uniprot_server_error_is_connection_error(monkeypatch): + monkeypatch.setattr( + uniprot.requests, "get", lambda url: FakeResponse(status_code=503) + ) + + with pytest.raises(ConnectionError): + fetch_uniprot("P07327") + + +def test_fetch_uniprot_inactive_entry(monkeypatch): + deleted = { + "entryType": "Inactive", + "primaryAccession": "A0A008APQ8", + "inactiveReason": { + "inactiveReasonType": "DELETED", + "deletedReason": "Not part of a reference proteome", + }, + } + monkeypatch.setattr(uniprot.requests, "get", lambda url: FakeResponse(deleted)) + + with pytest.raises(ValueError, match="inactive \\(DELETED: Not part of"): + fetch_uniprot("A0A008APQ8") + + +def test_fetch_uniprot_strips_whitespace(monkeypatch): + urls = [] + monkeypatch.setattr( + uniprot.requests, "get", lambda url: urls.append(url) or FakeResponse() + ) + + fetch_uniprot(" Q7DDU0\n") + + assert urls == ["https://rest.uniprot.org/uniprotkb/Q7DDU0.json"]