Make Input, Output, ModelSpec instances immutable.

HISTORY.rst

@@ -66,18 +66,22 @@
 Unreleased Changes
 ------------------
 
+General
+=======
+* To ensure consistent raster plot sizing in reports, padding is now added to
+  the top of raster plots if an adjacent raster plot has a units subheading.
+  (`#2471 <https://github.com/natcap/invest/issues/2471>`_)
+* ``Input, Output, and ModelSpec`` classes now create immutable objects.
+  Use ``model_copy(update=dict(...))`` to copy an object and update attributes.
+  (`#2228 <https://github.com/natcap/invest/issues/2228>`_)
+
 Workbench
 =========
 * Fixed a bug in Workbench application logs where the label referring to the
   origin of the log message was inaccurate. Messages are now labeled as from
   either the "main" or "renderer" process.
   (`#2522 <https://github.com/natcap/invest/issues/2522>`_)
 
-General
-=======
-* To ensure consistent raster plot sizing in reports, padding is now added to
-  the top of raster plots if an adjacent raster plot has a units subheading.
-  (`#2471 <https://github.com/natcap/invest/issues/2471>`_)
 
 3.19.0 (2026-04-16)
 -------------------

src/natcap/invest/spec.py

@@ -180,17 +180,28 @@ def validate_permissions_string(permissions):
     return permissions
 
 
-class Input(BaseModel):
+class ImmutableBaseModel(BaseModel):
+    """BaseModel with frozen attributes."""
+
+    model_config = ConfigDict(frozen=True)
+    """Make models immutable so that they must be copied before modifying."""
+
+
+class IOModel(ImmutableBaseModel):
+    """Base class for both `Input` and `Output`."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    """Allow fields to have arbitrary types that don't inherit from BaseModel
+    (needed for pint.Unit)."""
+
+
+class Input(IOModel):
     """A data input, or parameter, of an invest model.
 
     This represents an abstract input or parameter, which is rendered as an
     input field in the InVEST workbench. This does not store the value of the
     parameter for a specific run of the model.
     """
-    model_config = ConfigDict(arbitrary_types_allowed=True)
-    """Allow fields to have arbitrary types (that don't inherit from BaseModel).
-    Needed for pint.Unit."""
-
     id: str
     """Input identifier that should be unique within a model"""
 
@@ -236,7 +247,6 @@ def format_required_string(self) -> str:
             # assume that the about text will describe the conditional
             return gettext('conditionally required')
 
-
     def capitalize_name(self) -> str:
         """Capitalize a self.name into title case.
 
@@ -294,17 +304,13 @@ def describe_rst(self):
         return [rst_line]
 
 
-class Output(BaseModel):
+class Output(IOModel):
     """A data output, or result, of an invest model.
 
     This represents an abstract output which is produced as a result of running
     an invest model. This does not store the value of the output for a specific
     run of the model.
     """
-    model_config = ConfigDict(arbitrary_types_allowed=True)
-    """Allow fields to have arbitrary types (that don't inherit from BaseModel).
-    Needed for pint.Unit."""
-
     id: str
     """Output identifier that should be unique within a model"""
 
@@ -449,16 +455,8 @@ def format_path(p):
         return col.apply(format_path).astype(pandas.StringDtype())
 
 
-class RasterBand(BaseModel):
-    """A single-band raster input, or parameter, of an invest model.
-
-    This represents a raster file input (all GDAL-supported raster file types
-    are allowed), where only the first band is needed.
-    """
-    model_config = ConfigDict(arbitrary_types_allowed=True)
-    """Allow fields to have arbitrary types (that don't inherit from BaseModel).
-    Needed for pint.Unit."""
-
+class RasterBand(IOModel):
+    """A representation of a single raster band."""
     band_id: typing.Union[int, str] = 1
     """band index used to access the raster band"""
 
@@ -970,11 +968,16 @@ def get_validated_dataframe(self, csv_path: str, read_csv_kwargs={}, args=None):
 
         # Evaluate any conditional requirement strings to booleans
         # This will raise an error if any can't be evaluated
-        columns = copy.deepcopy(self.columns)
-        for col_spec in columns:
+        # Specs cannot be modified in place, so we copy with update
+        # to update the required attribute, and assemble a new list.
+        columns = []
+        for col_spec in self.columns:
             if isinstance(col_spec.required, str):
-                col_spec.required = bool(utils.evaluate_expression(
+                is_required = bool(utils.evaluate_expression(
                     col_spec.required, args or {}))
+                col_spec = col_spec.model_copy(update=dict(
+                    required=is_required))
+            columns.append(col_spec)
 
         for col_spec, pattern in zip(columns, patterns):
             matching_cols = [c for c in available_cols if re.fullmatch(pattern, c)]
@@ -1008,11 +1011,11 @@ def check_value(value):
                 df[col][df[col].notna()].apply(check_value)
 
         if any(df.columns.duplicated()):
-            duplicated_columns = df.columns[df.columns.duplicated]
+            duplicated_columns = df.columns[df.columns.duplicated()]
             raise ValueError(validation_messages.DUPLICATE_HEADER.format(
-                header=header_type,
-                header_name=expected,
-                number=count))
+                header=f'{self.orientation}(s)',
+                header_name=', '.join(duplicated_columns),
+                number='multiple'))
 
         # set the index column, if specified
         if self.index_col is not None:
@@ -1571,7 +1574,7 @@ def preprocess(self, value):
         return value
 
 
-class Option(BaseModel):
+class Option(ImmutableBaseModel):
     """An option in an OptionStringInput or OptionStringOutput."""
 
     key: str
@@ -1890,7 +1893,7 @@ class OptionStringOutput(Output):
     """A list of the values that this input may take"""
 
 
-class ModelSpec(BaseModel):
+class ModelSpec(ImmutableBaseModel):
     """Specification of an invest model describing metadata, inputs, and outputs."""
 
     model_id: str

src/natcap/invest/validation.py

@@ -1,20 +1,14 @@
 """Common validation utilities for InVEST models."""
-import copy
-import functools
 import importlib
 import inspect
 import logging
-import os
 import pprint
 
 import numpy
-import pint
 import pygeoprocessing
 from osgeo import gdal
-from osgeo import ogr
 from osgeo import osr
 
-from . import gettext
 from . import utils
 from . import validation_messages
 
@@ -221,22 +215,25 @@ def validate(args, model_spec):
         # Extra args that don't exist in the MODEL_SPEC are okay
         # we don't need to try to validate them
         try:
-            # Using deepcopy to make sure we don't modify the original spec
-            parameter_spec = copy.deepcopy(model_spec.get_input(key))
+            # Using a deep copy to make sure we don't modify the original spec
+            # or original nested specs.
+            parameter_spec = model_spec.get_input(key).model_copy(deep=True)
         except KeyError:
             LOGGER.debug(f'Provided key {key} does not exist in MODEL_SPEC')
             continue
 
         # rewrite parameter_spec for any nested, conditional validity
-        axis_keys = set(dir(parameter_spec)).intersection({'columns', 'rows', 'fields', 'contents'})
+        axis_keys = set(dir(parameter_spec)).intersection({'columns', 'fields', 'contents'})
 
         if axis_keys:
             for axis_key in axis_keys:
                 if getattr(parameter_spec, axis_key) is None:
                     continue
                 for nested_spec in getattr(parameter_spec, axis_key):
                     if (isinstance(nested_spec.required, str)):
-                        nested_spec.required = (
+                        # attributes have faux immutability; we cannot assign
+                        # to them, but we can assign to the underlying dict.
+                        nested_spec.__dict__['required'] = (
                             bool(utils.evaluate_expression(
                                 nested_spec.required, expression_values)))
         try:

tests/test_spec.py

@@ -366,8 +366,8 @@ def test_spec_no_results_suffix(self):
         }
 
         # Create LULC raster and pools csv in workspace and add them to args.
-        args['lulc_bas_path'] = os.path.join(args['workspace_dir'],
-                                       'lulc_bas_path.tif')
+        args['lulc_bas_path'] = os.path.join(
+            args['workspace_dir'], 'lulc_bas_path.tif')
         srs = osr.SpatialReference()
         srs.ImportFromEPSG(26910)  # UTM Zone 10N
         projection_wkt = srs.ExportToWkt()
@@ -492,6 +492,11 @@ def test_option_string_input_preprocess(self):
         self.assertEqual(option_string_input.preprocess(''), None)
         self.assertEqual(option_string_input.preprocess(None), None)
 
+    def test_immutable_input(self):
+        """Test that Input instances are immutable."""
+        with self.assertRaises(ValidationError):
+            spec.LULC.about = 'new description'
+
 
 class ModelSpecTests(unittest.TestCase):
     """Tests for natcap.invest.spec.ModelSpec."""
@@ -515,3 +520,9 @@ def test_reporter_field_validator(self):
         with self.assertRaises(ValidationError):
             # This module is importable, but has no 'report' attribute
             spec.ModelSpec(**data, reporter='natcap.invest')
+
+    def test_immutable_model_spec(self):
+        """Test that ModelSpec instance is immutable."""
+        from natcap.invest.carbon import MODEL_SPEC
+        with self.assertRaises(ValidationError):
+            MODEL_SPEC.model_id = 'foo'