Implement IXMP4Backend

.pre-commit-config.yaml

@@ -13,6 +13,7 @@ repos:
     - mypy >= 1.8.0
     - genno
     - GitPython
+    - ixmp4
     - nbclient
     - pytest
     - sphinx

ixmp/__init__.py

@@ -4,6 +4,7 @@
 
 from ixmp._config import config
 from ixmp.backend import BACKENDS, IAMC_IDX, ItemType
+from ixmp.backend.ixmp4 import IXMP4Backend
 from ixmp.backend.jdbc import JDBCBackend
 from ixmp.core.platform import Platform
 from ixmp.core.scenario import Scenario, TimeSeries
@@ -47,6 +48,7 @@
 
 # Register Backends provided by ixmp
 BACKENDS["jdbc"] = JDBCBackend
+BACKENDS["ixmp4"] = IXMP4Backend
 
 # Register Models provided by ixmp
 MODELS.update(

ixmp/backend/ixmp4.py

@@ -0,0 +1,91 @@
+from typing import TYPE_CHECKING
+
+from ixmp.backend.base import CachingBackend
+
+if TYPE_CHECKING:
+    import ixmp4
+
+
+class IXMP4Backend(CachingBackend):
+    """Backend using :mod:`ixmp4`."""
+
+    _platform: "ixmp4.Platform"
+
+    def __init__(self):
+        import ixmp4
+
+        # TODO Obtain `name` from the ixmp.Platform creating this Backend
+        name = "test"
+
+        # Add an ixmp4.Platform using ixmp4's own configuration code
+        # TODO Move this to a test fixture
+        # NB ixmp.tests.conftest.test_sqlite_mp exists, but is not importable (missing
+        #    __init__.py)
+        import ixmp4.conf
+
+        dsn = "sqlite:///:memory:"
+        try:
+            ixmp4.conf.settings.toml.add_platform(name, dsn)
+        except ixmp4.core.exceptions.PlatformNotUnique:
+            pass
+
+        # Instantiate and store
+        self._platform = ixmp4.Platform(name)
+
+    def get_scenarios(self, default, model, scenario):
+        # Current fails with:
+        # sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: run
+        # [SQL: SELECT DISTINCT run.model__id, run.scenario__id, run.version,
+        # run.is_default, run.id
+        # FROM run
+        # WHERE run.is_default = 1 ORDER BY run.id ASC]
+        return self._platform.runs.list()
+
+    # The below methods of base.Backend are not yet implemented
+    def _ni(self, *args, **kwargs):
+        raise NotImplementedError
+
+    add_model_name = _ni
+    add_scenario_name = _ni
+    cat_get_elements = _ni
+    cat_list = _ni
+    cat_set_elements = _ni
+    check_out = _ni
+    clear_solution = _ni
+    clone = _ni
+    commit = _ni
+    delete = _ni
+    delete_geo = _ni
+    delete_item = _ni
+    delete_meta = _ni
+    discard_changes = _ni
+    get = _ni
+    get_data = _ni
+    get_doc = _ni
+    get_geo = _ni
+    get_meta = _ni
+    get_model_names = _ni
+    get_nodes = _ni
+    get_scenario_names = _ni
+    get_timeslices = _ni
+    get_units = _ni
+    has_solution = _ni
+    init = _ni
+    init_item = _ni
+    is_default = _ni
+    item_delete_elements = _ni
+    item_get_elements = _ni
+    item_index = _ni
+    item_set_elements = _ni
+    last_update = _ni
+    list_items = _ni
+    remove_meta = _ni
+    run_id = _ni
+    set_as_default = _ni
+    set_data = _ni
+    set_doc = _ni
+    set_geo = _ni
+    set_meta = _ni
+    set_node = _ni
+    set_timeslice = _ni
+    set_unit = _ni

ixmp/tests/core/test_platform.py

@@ -2,6 +2,7 @@
 import logging
 import re
 from sys import getrefcount
+from typing import Generator
 from weakref import getweakrefcount
 
 import pandas as pd
@@ -15,21 +16,48 @@
 
 
 class TestPlatform:
-    def test_init(self):
+    @pytest.fixture(params=list(ixmp.BACKENDS))
+    def mp(self, request, test_mp) -> Generator[ixmp.Platform, None, None]:
+        """Fixture that yields 2 different platforms: one JDBC-backed, one ixmp4."""
+        backend = request.param
+
+        if backend == "jdbc":
+            yield test_mp
+        elif backend == "ixmp4":
+            # TODO Use a fixture similar to test_mp (with same contents) backed by ixmp4
+            yield ixmp.Platform(backend="ixmp4")
+
+    def test_init0(self):
         with pytest.raises(
-            ValueError, match=re.escape("backend class 'foo' not among ['jdbc']")
+            ValueError,
+            match=re.escape("backend class 'foo' not among ['ixmp4', 'jdbc']"),
         ):
             ixmp.Platform(backend="foo")
 
         # name="default" is used, referring to "local"
         mp = ixmp.Platform()
         assert "local" == mp.name
 
+    @pytest.mark.parametrize(
+        "backend, backend_args",
+        (
+            ("jdbc", dict(driver="hsqldb", url="jdbc:hsqldb:mem:TestPlatform")),
+            ("ixmp4", dict()),
+        ),
+    )
+    def test_init1(self, backend, backend_args):
+        # Platform can be instantiated
+        ixmp.Platform(backend=backend, **backend_args)
+
     def test_getattr(self, test_mp):
         """Test __getattr__."""
         with pytest.raises(AttributeError):
             test_mp.not_a_direct_backend_method
 
+    def test_scenario_list(self, mp):
+        scenario = mp.scenario_list()
+        assert isinstance(scenario, pd.DataFrame)
+
 
 @pytest.fixture
 def log_level_mp(test_mp):

pyproject.toml

@@ -56,10 +56,11 @@ docs = [
   "sphinx_rtd_theme",
   "sphinxcontrib-bibtex",
 ]
+ixmp4 = ["ixmp4"]
 report = ["genno[compat,graphviz]"]
 tutorial = ["jupyter"]
 tests = [
-  "ixmp[report,tutorial]",
+  "ixmp[ixmp4,report,tutorial]",
   "memory_profiler",
   "nbclient >= 0.5",
   "pretenders >= 1.4.4",