Introduce optimization variable

.github/workflows/pytest.yaml

@@ -39,12 +39,12 @@ jobs:
             postgres-version: "16"
             backend: "sqlite,rest-sqlite"
             pandas-version: "2.1.3"
-          # pandas 2.0.0
+          # pandas 2.1.0
           - python-version: "3.10"
             with-pyarrow: true
             postgres-version: "16"
             backend: "sqlite,rest-sqlite"
-            pandas-version: "2.0.0"
+            pandas-version: "2.1.0"
 
     name: py${{ matrix.python-version }} | backend=${{ matrix.backend }} | with-pyarrow=${{ matrix.with-pyarrow }} | pgsql=${{ matrix.postgres-version }} | pandas=${{ matrix.pandas-version }}
     runs-on: ubuntu-latest

ixmp4/core/__init__.py

@@ -4,6 +4,9 @@
 from .optimization.indexset import IndexSet as IndexSet
 from .optimization.scalar import Scalar as Scalar
 from .optimization.table import Table as Table
+
+# TODO Is this really the name we want to use?
+from .optimization.variable import Variable as OptimizationVariable
 from .platform import Platform as Platform
 from .region import Region as Region
 from .run import Run as Run

ixmp4/core/exceptions.py

@@ -186,9 +186,6 @@ class OptimizationDataValidationError(IxmpError):
     http_error_name = "optimization_data_validation_error"
 
 
-# == Optimization.Table ==
-
-
-class OptimizationTableUsageError(IxmpError):
+class OptimizationItemUsageError(IxmpError):
     http_status_code = 422
-    http_error_name = "optimization_table_usage_error"
+    http_error_name = "optimization_item_usage_error"

ixmp4/core/optimization/data.py

@@ -4,6 +4,7 @@
 from .indexset import IndexSetRepository
 from .scalar import ScalarRepository
 from .table import TableRepository
+from .variable import VariableRepository
 
 
 class OptimizationData(BaseFacade):
@@ -13,9 +14,11 @@ class OptimizationData(BaseFacade):
     indexsets: IndexSetRepository
     scalars: ScalarRepository
     tables: TableRepository
+    variables: VariableRepository
 
     def __init__(self, *args, run: Run, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self.indexsets = IndexSetRepository(_backend=self.backend, _run=run)
         self.scalars = ScalarRepository(_backend=self.backend, _run=run)
         self.tables = TableRepository(_backend=self.backend, _run=run)
+        self.variables = VariableRepository(_backend=self.backend, _run=run)

ixmp4/core/optimization/variable.py

@@ -0,0 +1,144 @@
+from datetime import datetime
+from typing import Any, ClassVar, Iterable
+
+import pandas as pd
+
+from ixmp4.core.base import BaseFacade, BaseModelFacade
+from ixmp4.data.abstract import Docs as DocsModel
+from ixmp4.data.abstract import OptimizationVariable as VariableModel
+from ixmp4.data.abstract import Run
+from ixmp4.data.abstract.optimization import Column
+
+
+class Variable(BaseModelFacade):
+    _model: VariableModel
+    NotFound: ClassVar = VariableModel.NotFound
+    NotUnique: ClassVar = VariableModel.NotUnique
+
+    @property
+    def id(self) -> int:
+        return self._model.id
+
+    @property
+    def name(self) -> str:
+        return self._model.name
+
+    @property
+    def run_id(self) -> int:
+        return self._model.run__id
+
+    @property
+    def data(self) -> dict[str, Any]:
+        return self._model.data
+
+    def add(self, data: dict[str, Any] | pd.DataFrame) -> None:
+        """Adds data to an existing Variable."""
+        self.backend.optimization.variables.add_data(
+            variable_id=self._model.id, data=data
+        )
+        self._model.data = self.backend.optimization.variables.get(
+            run_id=self._model.run__id, name=self._model.name
+        ).data
+
+    def remove_data(self) -> None:
+        """Removes data from an existing Variable."""
+        self.backend.optimization.variables.remove_data(variable_id=self._model.id)
+        self._model.data = self.backend.optimization.variables.get(
+            run_id=self._model.run__id, name=self._model.name
+        ).data
+
+    @property
+    def levels(self) -> list:
+        return self._model.data.get("levels", [])
+
+    @property
+    def marginals(self) -> list:
+        return self._model.data.get("marginals", [])
+
+    @property
+    def constrained_to_indexsets(self) -> list[str]:
+        return (
+            [column.indexset.name for column in self._model.columns]
+            if self._model.columns
+            else []
+        )
+
+    @property
+    def columns(self) -> list[Column] | None:
+        return self._model.columns
+
+    @property
+    def created_at(self) -> datetime | None:
+        return self._model.created_at
+
+    @property
+    def created_by(self) -> str | None:
+        return self._model.created_by
+
+    @property
+    def docs(self):
+        try:
+            return self.backend.optimization.variables.docs.get(self.id).description
+        except DocsModel.NotFound:
+            return None
+
+    @docs.setter
+    def docs(self, description):
+        if description is None:
+            self.backend.optimization.variables.docs.delete(self.id)
+        else:
+            self.backend.optimization.variables.docs.set(self.id, description)
+
+    @docs.deleter
+    def docs(self):
+        try:
+            self.backend.optimization.variables.docs.delete(self.id)
+        # TODO: silently failing
+        except DocsModel.NotFound:
+            return None
+
+    def __str__(self) -> str:
+        return f"<Variable {self.id} name={self.name}>"
+
+
+class VariableRepository(BaseFacade):
+    _run: Run
+
+    def __init__(self, _run: Run, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._run = _run
+
+    def create(
+        self,
+        name: str,
+        constrained_to_indexsets: str | list[str] | None = None,
+        column_names: list[str] | None = None,
+    ) -> Variable:
+        model = self.backend.optimization.variables.create(
+            name=name,
+            run_id=self._run.id,
+            constrained_to_indexsets=constrained_to_indexsets,
+            column_names=column_names,
+        )
+        return Variable(_backend=self.backend, _model=model)
+
+    def get(self, name: str) -> Variable:
+        model = self.backend.optimization.variables.get(run_id=self._run.id, name=name)
+        return Variable(_backend=self.backend, _model=model)
+
+    def list(self, name: str | None = None) -> Iterable[Variable]:
+        variables = self.backend.optimization.variables.list(
+            run_id=self._run.id, name=name
+        )
+        return [
+            Variable(
+                _backend=self.backend,
+                _model=i,
+            )
+            for i in variables
+        ]
+
+    def tabulate(self, name: str | None = None) -> pd.DataFrame:
+        return self.backend.optimization.variables.tabulate(
+            run_id=self._run.id, name=name
+        )

ixmp4/data/abstract/__init__.py

@@ -37,6 +37,10 @@
     Table,
     TableRepository,
 )
+
+# TODO for PR: avoiding name conflict here Is that okay?
+from .optimization import Variable as OptimizationVariable
+from .optimization import VariableRepository as OptimizationVariableRepository
 from .region import Region, RegionRepository
 from .run import Run, RunRepository
 from .scenario import Scenario, ScenarioRepository

ixmp4/data/abstract/optimization/__init__.py

@@ -2,3 +2,4 @@
 from .indexset import IndexSet, IndexSetRepository
 from .scalar import Scalar, ScalarRepository
 from .table import Table, TableRepository
+from .variable import Variable, VariableRepository

ixmp4/data/abstract/optimization/column.py

@@ -14,8 +14,10 @@ class Column(base.BaseModel, Protocol):
     """Unique name of the Column."""
     dtype: types.String
     """Type of the Column's data."""
-    table__id: types.Integer
+    table__id: types.Mapped[int | None]
     """Foreign unique integer id of a Table."""
+    variable__id: types.Mapped[int | None]
+    """Foreign unique integer id of a Variable."""
     indexset: types.Mapped[IndexSet]
     """Associated IndexSet."""
     constrained_to_indexset: types.Integer