feat: add embl gems repository

release-notes/next-release.md

@@ -3,6 +3,7 @@
 ## New features
 
 * View number of genes in model notebook representation.
+* Add EMBL GEMs (https://github.com/cdanielmachado/embl_gems) to the list of repositories.
 
 ## Fixes
 

src/cobra/io/__init__.py

@@ -6,4 +6,10 @@
 from cobra.io.mat import load_matlab_model, save_matlab_model
 from cobra.io.sbml import read_sbml_model, write_sbml_model, validate_sbml_model
 from cobra.io.yaml import from_yaml, load_yaml_model, save_yaml_model, to_yaml
-from cobra.io.web import AbstractModelRepository, BiGGModels, BioModels, load_model
+from cobra.io.web import (
+    AbstractModelRepository,
+    BiGGModels,
+    BioModels,
+    EMBLGems,
+    load_model,
+)

src/cobra/io/web/__init__.py

@@ -4,4 +4,5 @@
 from .abstract_model_repository import AbstractModelRepository
 from .bigg_models_repository import BiGGModels
 from .biomodels_repository import BioModels
+from .embl_gems_repository import EMBLGems
 from .load import load_model

src/cobra/io/web/bigg_models_repository.py

@@ -1,4 +1,4 @@
-"""Provide a concrete implementation of the BioModels repository interface."""
+"""Provide a concrete implementation of the BiGG repository interface."""
 
 
 from io import BytesIO

src/cobra/io/web/embl_gems_repository.py

@@ -0,0 +1,98 @@
+"""
+Provide a concrete implementation of the carveme repository interface.
+"""
+
+
+from io import BytesIO
+
+import httpx
+
+from .abstract_model_repository import AbstractModelRepository
+
+
+def _decode_model_path(model_path):
+    """Decode the model path to EMBL GEMs."""
+    tokens = model_path.split("_")
+    genus = tokens[0]
+
+    directory = genus.lower()
+    alphabet = directory[0]
+
+    return f"{alphabet}/{directory}/{model_path}"
+
+
+class EMBLGems(AbstractModelRepository):
+    """
+    Define a concrete implementation of the EMBL GEMs repository.
+
+    Attributes
+    ----------
+    name : str
+        The name of the EMBL GEMs repository.
+
+    """
+
+    name: str = "EMBL GEMs"
+
+    def __init__(
+        self,
+        **kwargs,
+    ) -> None:
+        """
+        Initialize a EMBL GEMs repository interface.
+
+        Other Parameters
+        ----------------
+        kwargs
+            Passed to the parent constructor in order to enable multiple inheritance.
+
+        """
+        super().__init__(
+            url="https://github.com/cdanielmachado/embl_gems/blob/master/models/",
+            **kwargs,
+        )
+
+    def get_sbml(self, model_id: str) -> bytes:
+        """
+        Attempt to download an SBML document from the repository.
+
+        Parameters
+        ----------
+        model_id : str
+            The identifier of the desired metabolic model. This is typically repository
+            specific.
+
+        Returns
+        -------
+        bytes
+            A gzip-compressed, UTF-8 encoded SBML document.
+
+        Raises
+        ------
+        httpx.HTTPError
+            In case there are any connection problems.
+
+        """
+        compressed = BytesIO()
+
+        decoded_path = _decode_model_path(model_id)
+
+        filename = f"{model_id}.xml.gz"
+        print(self._url.join(decoded_path).join(filename))
+        with self._progress, httpx.stream(
+            method="GET",
+            url=self._url.join(decoded_path).join(filename),
+            params={"raw": "true"},
+            follow_redirects=True,
+        ) as response:
+            response.raise_for_status()
+            task_id = self._progress.add_task(
+                description="download",
+                total=int(response.headers["Content-Length"]),
+                model_id=model_id,
+            )
+            for chunk in response.iter_bytes():
+                compressed.write(chunk)
+                self._progress.update(task_id=task_id, advance=len(chunk))
+        compressed.seek(0)
+        return compressed.read()

src/cobra/io/web/load.py

@@ -15,6 +15,7 @@
 from .bigg_models_repository import BiGGModels
 from .biomodels_repository import BioModels
 from .cobrapy_repository import Cobrapy
+from .embl_gems_repository import EMBLGems
 
 
 if TYPE_CHECKING:
@@ -29,6 +30,7 @@
     Cobrapy(),
     BiGGModels(),
     BioModels(),
+    EMBLGems()
 )
 
 

tests/test_io/test_web/test_load.py

@@ -8,7 +8,7 @@
 import pytest
 
 from cobra import Configuration
-from cobra.io import BiGGModels, BioModels, load_model
+from cobra.io import BiGGModels, BioModels, EMLBGems, load_model
 
 
 if TYPE_CHECKING:
@@ -39,6 +39,12 @@ def biomodels(mini_sbml: bytes, mocker: "MockerFixture") -> Mock:
     result.get_sbml.return_value = mini_sbml
     return result
 
+@pytest.fixture
+def embl_gems(mini_sbml: bytes, mocker: "MockerFixture") -> Mock:
+    """Provide a mocked EMBL Gems repository interface."""
+    result = mocker.Mock(spec_set=EMLBGems)
+    result.get_sbml.return_value = mini_sbml
+    return result
 
 def test_bigg_access(bigg_models: Mock) -> None:
     """Test that SBML would be retrieved from the BiGG Models repository.
@@ -66,6 +72,30 @@ def test_biomodels_access(biomodels: Mock) -> None:
     biomodels.get_sbml.assert_called_once_with(model_id="BIOMD0000000633")
 
 
+def test_biomodels_access(biomodels: Mock) -> None:
+    """Test that SBML would be retrieved from the BioModels repository.
+
+    Parameters
+    ----------
+    biomodels : unittest.mock.Mock
+        The mocked object for BioModels model respository.
+
+    """
+    load_model("BIOMD0000000633", cache=False, repositories=[biomodels])
+    biomodels.get_sbml.assert_called_once_with(model_id="BIOMD0000000633")
+
+def test_biomodels_access(embl_gems: Mock) -> None:
+    """Test that SBML would be retrieved from the EMBL Gems repository.
+
+    Parameters
+    ----------
+    embl_gems : unittest.mock.Mock
+        The mocked object for BioModels model respository.
+
+    """
+    load_model("Abiotrophia_defectiva_ATCC_49176", cache=False, repositories=[embl_gems])
+    biomodels.get_sbml.assert_called_once_with(model_id="Abiotrophia_defectiva_ATCC_49176")
+
 def test_unknown_model() -> None:
     """Expect that a not found error is raised (e2e)."""
     with pytest.raises(RuntimeError):
@@ -75,6 +105,7 @@ def test_unknown_model() -> None:
 @pytest.mark.parametrize(
     "model_id, num_metabolites, num_reactions",
     [("e_coli_core", 72, 95), ("BIOMD0000000633", 50, 35)],
+    [("Abiotrophia_defectiva_ATCC_49176", 1070, 826)]
 )
 def test_remote_load(model_id: str, num_metabolites: int, num_reactions: int) -> None:
     """Test that sample models can be loaded from remote repositories (e2e).