add iamc filter for tabulate/ list runs

ixmp4/data/db/run/filter.py

@@ -1,7 +1,27 @@
 from ixmp4.data.db import filters as base
-from ixmp4.db import filters
+from ixmp4.data.db.iamc.datapoint import get_datapoint_model
+from ixmp4.data.db.iamc.timeseries import TimeSeries
+from ixmp4.data.db.run.model import Run
+from ixmp4.db import filters, utils
+
+
+class IamcRunFilter(filters.BaseFilter, metaclass=filters.FilterMeta):
+    region: base.RegionFilter
+    variable: base.VariableFilter
+    unit: base.UnitFilter
+
+    def join(self, exc, session=None):
+        if not utils.is_joined(exc, TimeSeries):
+            exc = exc.join(TimeSeries, onclause=TimeSeries.run__id == Run.id)
+
+        model = get_datapoint_model(session)
+        if not utils.is_joined(exc, model):
+            exc = exc.join(model, onclause=model.time_series__id == TimeSeries.id)
+        return exc
 
 
 class RunFilter(base.RunFilter, metaclass=filters.FilterMeta):
+    iamc: IamcRunFilter | filters.Boolean
+
     def join(self, exc, **kwargs):
         return exc

tests/core/test_run.py

@@ -4,7 +4,7 @@
 
 from ixmp4 import IxmpError, Run
 
-from ..utils import all_platforms
+from ..utils import all_platforms, create_filter_test_data
 
 
 def _expected_runs_table(*row_default):
@@ -93,3 +93,63 @@ def test_run_versions(self, test_mp, request):
         # non-default version cannot be again set as un-default
         with pytest.raises(IxmpError):
             run2.unset_as_default()
+
+    def test_filter_run(self, test_mp, request):
+        test_mp = request.getfixturevalue(test_mp)
+        run1, run2 = create_filter_test_data(test_mp)
+
+        res = test_mp.runs.tabulate(
+            default_only=False,
+            iamc={
+                "region": {"name": "Region 1"},
+            },
+        )
+        assert sorted(res["model"].tolist()) == ["Model 1"]
+        assert sorted(res["scenario"].tolist()) == ["Scenario 1"]
+
+        res = test_mp.runs.tabulate(
+            default_only=False,
+            iamc={
+                "region": {"name": "Region 3"},
+            },
+        )
+        assert sorted(res["model"].tolist()) == ["Model 1", "Model 1", "Model 2"]
+        assert sorted(res["scenario"].tolist()) == [
+            "Scenario 1",
+            "Scenario 1",
+            "Scenario 2",
+        ]
+
+        run1.set_as_default()
+        res = test_mp.runs.tabulate(
+            iamc={
+                "variable": {"name": "Variable 1"},
+                "unit": {"name__in": ["Unit 3", "Unit 4"]},
+            }
+        )
+        assert sorted(res["model"].tolist()) == ["Model 2"]
+        assert sorted(res["scenario"].tolist()) == ["Scenario 2"]
+
+        res = test_mp.runs.tabulate(
+            default_only=False,
+            scenario={"name": "Scenario 2"},
+        )
+        assert sorted(res["model"].tolist()) == ["Model 2"]
+        assert sorted(res["scenario"].tolist()) == ["Scenario 2"]
+
+        remove_data = test_mp.iamc.tabulate(
+            model={"name": "Model 1"},
+            scenario={"name": "Scenario 1"},
+            run={"default_only": False},
+            region={"name": "Region 3"},
+        )
+        run1.iamc.remove(remove_data)
+        run2.iamc.remove(remove_data)
+        res = test_mp.runs.tabulate(
+            default_only=False,
+            iamc={
+                "region": {"name": "Region 3"},
+            },
+        )
+        assert sorted(res["model"].tolist()) == ["Model 2"]
+        assert sorted(res["scenario"].tolist()) == ["Scenario 2"]