feat(dfn): add from_dict method (#241)

Allows using Dfn/Field spec objects explicitly by passing a dictionary to __init__ via double star syntax, or via from_dict which handles field structuring automatically, and with strict=False ignores unrecognized keys (like pydantic's extra="ignore"), if strict=True unrecognized keys cause an error (like pydantic's extra="forbid")

Author: wpbonelli

autotest/test_dfn.py

@@ -1,9 +1,13 @@
+from dataclasses import asdict
 from pathlib import Path
 
 import pytest
+from packaging.version import Version
 
-from modflow_devtools.dfn import _load_common, load, load_flat
+from modflow_devtools.dfn import Dfn, _load_common, load, load_flat
 from modflow_devtools.dfn.fetch import fetch_dfns
+from modflow_devtools.dfn.schema.v1 import FieldV1
+from modflow_devtools.dfn.schema.v2 import FieldV2
 from modflow_devtools.dfn2toml import convert
 from modflow_devtools.markers import requires_pkg
 
@@ -114,3 +118,233 @@ def test_convert(function_tmpdir):
             assert dis.parent == "gwf"
             assert "options" in (dis.blocks or {})
             assert "dimensions" in (dis.blocks or {})
+
+
+def test_dfn_from_dict_ignores_extra_keys():
+    d = {
+        "schema_version": Version("2"),
+        "name": "test-dfn",
+        "extra_key": "should be allowed",
+        "another_extra": 123,
+    }
+    dfn = Dfn.from_dict(d)
+    assert dfn.name == "test-dfn"
+    assert dfn.schema_version == Version("2")
+
+
+def test_dfn_from_dict_strict_mode():
+    d = {
+        "schema_version": Version("2"),
+        "name": "test-dfn",
+        "extra_key": "should cause error",
+    }
+    with pytest.raises(ValueError, match="Unrecognized keys in DFN data"):
+        Dfn.from_dict(d, strict=True)
+
+
+def test_dfn_from_dict_strict_mode_nested():
+    d = {
+        "schema_version": Version("2"),
+        "name": "test-dfn",
+        "blocks": {
+            "options": {
+                "test_field": {
+                    "name": "test_field",
+                    "type": "keyword",
+                    "extra_key": "should cause error",
+                },
+            },
+        },
+    }
+    with pytest.raises(ValueError, match="Unrecognized keys in field data"):
+        Dfn.from_dict(d, strict=True)
+
+
+def test_dfn_from_dict_roundtrip():
+    original = Dfn(
+        schema_version=Version("2"),
+        name="gwf-nam",
+        parent="sim-nam",
+        advanced=False,
+        multi=True,
+        blocks={"options": {}},
+    )
+    d = asdict(original)
+    reconstructed = Dfn.from_dict(d)
+    assert reconstructed.name == original.name
+    assert reconstructed.schema_version == original.schema_version
+    assert reconstructed.parent == original.parent
+    assert reconstructed.advanced == original.advanced
+    assert reconstructed.multi == original.multi
+    assert reconstructed.blocks == original.blocks
+
+
+def test_fieldv1_from_dict_ignores_extra_keys():
+    d = {
+        "name": "test_field",
+        "type": "keyword",
+        "extra_key": "should be allowed",
+        "another_extra": 123,
+    }
+    field = FieldV1.from_dict(d)
+    assert field.name == "test_field"
+    assert field.type == "keyword"
+
+
+def test_fieldv1_from_dict_strict_mode():
+    d = {
+        "name": "test_field",
+        "type": "keyword",
+        "extra_key": "should cause error",
+    }
+    with pytest.raises(ValueError, match="Unrecognized keys in field data"):
+        FieldV1.from_dict(d, strict=True)
+
+
+def test_fieldv1_from_dict_roundtrip():
+    original = FieldV1(
+        name="maxbound",
+        type="integer",
+        block="dimensions",
+        description="maximum number of cells",
+        tagged=True,
+    )
+    d = asdict(original)
+    reconstructed = FieldV1.from_dict(d)
+    assert reconstructed.name == original.name
+    assert reconstructed.type == original.type
+    assert reconstructed.block == original.block
+    assert reconstructed.description == original.description
+    assert reconstructed.tagged == original.tagged
+
+
+def test_fieldv2_from_dict_ignores_extra_keys():
+    d = {
+        "name": "test_field",
+        "type": "keyword",
+        "extra_key": "should be allowed",
+        "another_extra": 123,
+    }
+    field = FieldV2.from_dict(d)
+    assert field.name == "test_field"
+    assert field.type == "keyword"
+
+
+def test_fieldv2_from_dict_strict_mode():
+    d = {
+        "name": "test_field",
+        "type": "keyword",
+        "extra_key": "should cause error",
+    }
+    with pytest.raises(ValueError, match="Unrecognized keys in field data"):
+        FieldV2.from_dict(d, strict=True)
+
+
+def test_fieldv2_from_dict_roundtrip():
+    original = FieldV2(
+        name="nper",
+        type="integer",
+        block="dimensions",
+        description="number of stress periods",
+        optional=False,
+    )
+    d = asdict(original)
+    reconstructed = FieldV2.from_dict(d)
+    assert reconstructed.name == original.name
+    assert reconstructed.type == original.type
+    assert reconstructed.block == original.block
+    assert reconstructed.description == original.description
+    assert reconstructed.optional == original.optional
+
+
+def test_dfn_from_dict_with_v1_field_dicts():
+    d = {
+        "schema_version": Version("1"),
+        "name": "test-dfn",
+        "blocks": {
+            "options": {
+                "save_flows": {
+                    "name": "save_flows",
+                    "type": "keyword",
+                    "tagged": True,
+                    "in_record": False,
+                },
+            },
+        },
+    }
+    dfn = Dfn.from_dict(d)
+    assert dfn.schema_version == Version("1")
+    assert dfn.name == "test-dfn"
+    assert dfn.blocks is not None
+    assert "options" in dfn.blocks
+    assert "save_flows" in dfn.blocks["options"]
+
+    field = dfn.blocks["options"]["save_flows"]
+    assert isinstance(field, FieldV1)
+    assert field.name == "save_flows"
+    assert field.type == "keyword"
+    assert field.tagged is True
+    assert field.in_record is False
+
+
+def test_dfn_from_dict_with_v2_field_dicts():
+    d = {
+        "schema_version": Version("2"),
+        "name": "test-dfn",
+        "blocks": {
+            "dimensions": {
+                "nper": {
+                    "name": "nper",
+                    "type": "integer",
+                    "optional": False,
+                },
+            },
+        },
+    }
+    dfn = Dfn.from_dict(d)
+    assert dfn.schema_version == Version("2")
+    assert dfn.name == "test-dfn"
+    assert dfn.blocks is not None
+    assert "dimensions" in dfn.blocks
+    assert "nper" in dfn.blocks["dimensions"]
+
+    field = dfn.blocks["dimensions"]["nper"]
+    assert isinstance(field, FieldV2)
+    assert field.name == "nper"
+    assert field.type == "integer"
+    assert field.optional is False
+
+
+def test_dfn_from_dict_defaults_to_v2_fields():
+    d = {
+        "name": "test-dfn",
+        "blocks": {
+            "options": {
+                "some_field": {
+                    "name": "some_field",
+                    "type": "keyword",
+                },
+            },
+        },
+    }
+    dfn = Dfn.from_dict(d)
+    assert dfn.blocks is not None
+    field = dfn.blocks["options"]["some_field"]
+    assert isinstance(field, FieldV2)
+    assert dfn.schema_version == Version("2")
+
+
+def test_dfn_from_dict_with_already_deserialized_fields():
+    field = FieldV2(name="test", type="keyword")
+    d = {
+        "schema_version": Version("2"),
+        "name": "test-dfn",
+        "blocks": {
+            "options": {
+                "test": field,
+            },
+        },
+    }
+    dfn = Dfn.from_dict(d)
+    assert dfn.blocks is not None
+    assert dfn.blocks["options"]["test"] is field

modflow_devtools/dfn/__init__.py

@@ -24,7 +24,7 @@
     try_parse_bool,
     try_parse_parent,
 )
-from modflow_devtools.dfn.schema.block import Block, Blocks
+from modflow_devtools.dfn.schema.block import Block, Blocks, block_sort_key
 from modflow_devtools.dfn.schema.field import SCALAR_TYPES, Field, Fields
 from modflow_devtools.dfn.schema.ref import Ref
 from modflow_devtools.dfn.schema.v1 import FieldV1
@@ -42,6 +42,7 @@
     "FieldV2",
     "Fields",
     "Ref",
+    "block_sort_key",
     "load",
     "load_flat",
     "load_tree",
@@ -90,6 +91,52 @@ def fields(self) -> Fields:
         # TODO: change to normal dict after deprecating v1 schema
         return OMD(fields)
 
+    @classmethod
+    def from_dict(cls, d: dict, strict: bool = False) -> "Dfn":
+        """
+        Create a Dfn instance from a dictionary.
+
+        Parameters
+        ----------
+        d : dict
+            Dictionary containing DFN data
+        strict : bool, optional
+            If True, raise ValueError if dict contains unrecognized keys at the
+            top level or in nested field dicts. If False (default), ignore
+            unrecognized keys.
+        """
+        keys = list(cls.__annotations__.keys())
+        if strict:
+            extra_keys = set(d.keys()) - set(keys)
+            if extra_keys:
+                raise ValueError(f"Unrecognized keys in DFN data: {extra_keys}")
+        data = {k: v for k, v in d.items() if k in keys}
+        schema_version = data.get("schema_version", Version("2"))
+        field_cls = FieldV1 if schema_version == Version("1") else FieldV2
+
+        def _fields(block_name, block_data):
+            fields = {}
+            for field_name, field_data in block_data.items():
+                if isinstance(field_data, dict):
+                    fields[field_name] = field_cls.from_dict(field_data, strict=strict)
+                elif isinstance(field_data, field_cls):
+                    fields[field_name] = field_data
+                else:
+                    raise TypeError(
+                        f"Invalid field data for {field_name} in block {block_name}: "
+                        f"expected dict or Field, got {type(field_data)}"
+                    )
+            return fields
+
+        if blocks := data.get("blocks"):
+            data["schema_version"] = schema_version
+            data["blocks"] = {
+                block_name: _fields(block_name, block_data)
+                for block_name, block_data in blocks.items()
+            }
+
+        return cls(**data)
+
 
 class SchemaMap(ABC):
     @abstractmethod

modflow_devtools/dfn/schema/v1.py

@@ -17,7 +17,21 @@ class FieldV1(Field):
     mf6internal: str | None = None
 
     @classmethod
-    def from_dict(cls, d: dict) -> "FieldV1":
-        """Create a FieldV1 instance from a dictionary."""
+    def from_dict(cls, d: dict, strict: bool = False) -> "FieldV1":
+        """
+        Create a FieldV1 instance from a dictionary.
+
+        Parameters
+        ----------
+        d : dict
+            Dictionary containing field data
+        strict : bool, optional
+            If True, raise ValueError if dict contains unrecognized keys.
+            If False (default), ignore unrecognized keys.
+        """
         keys = list(cls.__annotations__.keys()) + list(Field.__annotations__.keys())
+        if strict:
+            extra_keys = set(d.keys()) - set(keys)
+            if extra_keys:
+                raise ValueError(f"Unrecognized keys in field data: {extra_keys}")
         return cls(**{k: v for k, v in d.items() if k in keys})

modflow_devtools/dfn/schema/v2.py

@@ -13,7 +13,21 @@ class FieldV2(Field):
     pass
 
     @classmethod
-    def from_dict(cls, d: dict) -> "FieldV2":
-        """Create a FieldV2 instance from a dictionary."""
+    def from_dict(cls, d: dict, strict: bool = False) -> "FieldV2":
+        """
+        Create a FieldV2 instance from a dictionary.
+
+        Parameters
+        ----------
+        d : dict
+            Dictionary containing field data
+        strict : bool, optional
+            If True, raise ValueError if dict contains unrecognized keys.
+            If False (default), ignore unrecognized keys.
+        """
         keys = list(cls.__annotations__.keys()) + list(Field.__annotations__.keys())
+        if strict:
+            extra_keys = set(d.keys()) - set(keys)
+            if extra_keys:
+                raise ValueError(f"Unrecognized keys in field data: {extra_keys}")
         return cls(**{k: v for k, v in d.items() if k in keys})