Merge pull request #38 from csdms/mcflugen/add-parameters-to-stage

Add parameters keyword to stage function

Author: mcflugen

src/model_metadata/api.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
+from typing import Any
+
+from model_metadata.errors import UnknownKeyError
 from model_metadata.model_setup import FileSystemLoader
 from model_metadata.model_setup import OldFileSystemLoader
 from model_metadata.modelmetadata import ModelMetadata
@@ -52,7 +56,10 @@ def query(model: str, var: str) -> ModelMetadata:
 
 
 def stage(
-    model: str, dest: str = ".", old_style_templates: bool = False
+    model: str,
+    dest: str = ".",
+    old_style_templates: bool = False,
+    parameters: dict[str, Any] | None = None,
 ) -> tuple[str, ...]:
     """Stage a model by setting up its input files.
 
@@ -62,18 +69,39 @@ def stage(
         The model is interpreted either as a path to a folder that
         contains metadata, the name of a model component, or a
         model object.
-    dest : str
+    dest : str, optional
         Path to a folder within which to stage the model.
+    parameters : dict[str, Any], optional
+        A dictionary of parameters that overrides the default
+        values.
     """
+    parameters = {} if parameters is None else parameters
+
     defaults = {}
     mmd = ModelMetadata.find(model)
     meta = ModelMetadata(mmd)
     for param, item in meta.parameters.items():
         defaults[param] = item["value"]["default"]
 
+    try:
+        _check_for_unknown_keys(defaults.keys(), parameters.keys())
+    except UnknownKeyError as e:
+        e.add_note(
+            f"valid parameter{'s' if len(defaults) > 1 else ''}:"
+            f" {', '.join(sorted(repr(k) for k in defaults))}"
+        )
+        raise
+
+    defaults = {**defaults, **parameters}
+
     if old_style_templates:
         manifest = OldFileSystemLoader(mmd).stage_all(dest, **defaults)
     else:
         manifest = FileSystemLoader(mmd).stage_all(dest, **defaults)
 
     return manifest
+
+
+def _check_for_unknown_keys(allowed: Iterable[str], user: Iterable[str]) -> None:
+    if unknown_keys := (set(user) - set(allowed)):
+        raise UnknownKeyError(unknown_keys)

src/model_metadata/errors.py

@@ -1,6 +1,8 @@
 #! /usr/bin/env python
 from __future__ import annotations
 
+from collections.abc import Iterable
+
 
 class ModelMetadataError(Exception):
     """Base error for model_metadata package."""
@@ -55,3 +57,16 @@ def __init__(self, entry_point: str, msg: str | None = None):
 
     def __str__(self) -> str:
         return self._entry_point + f": {self._msg}" if self._msg else ""
+
+
+class UnknownKeyError(ModelMetadataError):
+    """Raise if a dictionary contains one or more unrecognized keys."""
+
+    def __init__(self, unknown: Iterable[str]) -> None:
+        super().__init__(*sorted(set(unknown)))
+
+    def __str__(self) -> str:
+        return (
+            f"unknown key{'s' if len(self.args) > 1 else ''}:"
+            f" {', '.join(repr(key) for key in self.args)}"
+        )

tests/api_test.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import itertools
 import os
 import pathlib
 
@@ -10,6 +11,7 @@
 from model_metadata.errors import MetadataNotFoundError
 from model_metadata.errors import MissingSectionError
 from model_metadata.errors import MissingValueError
+from model_metadata.errors import UnknownKeyError
 
 
 class Model:
@@ -85,6 +87,26 @@ def test_stage_is_filled_out(tmpdir, shared_datadir):
         assert "}}" not in contents
 
 
+def test_stage_with_parameters(tmpdir, shared_datadir):
+    with tmpdir.as_cwd():
+        stage(str(shared_datadir), parameters={"run_duration": 999})
+
+        file_contains_runtime = False
+        with open("child.in") as fp:
+            for this, next_ in itertools.pairwise(fp):
+                file_contains_runtime = this.startswith("RUNTIME")
+                if file_contains_runtime:
+                    assert next_.startswith("999")
+                    break
+        assert file_contains_runtime
+
+
+@pytest.mark.parametrize("params", ({"foo": "bar"}, {"foo": "bar", "baz": "foobar"}))
+def test_stage_with_unknown_parameters(tmpdir, shared_datadir, params):
+    with tmpdir.as_cwd(), pytest.raises(UnknownKeyError):
+        stage(str(shared_datadir), parameters=params)
+
+
 def test_query_with_bad_section(shared_datadir):
     with pytest.raises(MissingSectionError):
         query(str(shared_datadir), "not-a-section.version")