Validate configs based on most primitive JSON types

docs/conf.py

@@ -21,7 +21,10 @@
 html_static_path = ["static"]
 html_theme = "sphinx_rtd_theme"
 intersphinx_mapping = {"python": ("https://docs.python.org/3", None)}
-nitpick_ignore_regex = [("py:class", r"^uwtools\..*"), ("py:class", "f90nml.Namelist")]
+nitpick_ignore = [
+    ("py:class", "Path"),
+    ("py:class", "f90nml.Namelist"),
+]
 numfig = True
 numfig_format = {"figure": "Figure %s"}
 project = "Unified Workflow Tools"

docs/sections/user_guide/api/config.rst

@@ -5,5 +5,5 @@
    :target: https://mybinder.org/v2/gh/ufs-community/uwtools/main?labpath=notebooks%2Fconfig.ipynb
 
 .. automodule:: uwtools.api.config
-   :inherited-members: UserDict
    :members:
+   :show-inheritance:

recipe/meta.json

@@ -36,5 +36,5 @@
       "requests =2.32.*"
     ]
   },
-  "version": "2.5.0"
+  "version": "2.6.0"
 }

src/uwtools/api/config.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Optional, Union
 
-from uwtools.config.formats.base import Config as _Config
+from uwtools.config.formats.base import Config
 from uwtools.config.formats.fieldtable import FieldTableConfig
 from uwtools.config.formats.ini import INIConfig
 from uwtools.config.formats.nml import NMLConfig
@@ -15,6 +15,8 @@
 from uwtools.config.support import YAMLKey
 from uwtools.config.tools import compare_configs as _compare
 from uwtools.config.tools import realize_config as _realize
+from uwtools.config.validator import ConfigDataT, ConfigPathT
+from uwtools.config.validator import validate_check_config as _validate_check_config
 from uwtools.config.validator import validate_external as _validate_external
 from uwtools.exceptions import UWConfigError
 from uwtools.utils.api import ensure_data_source as _ensure_data_source
@@ -111,9 +113,9 @@ def get_yaml_config(
 
 
 def realize(
-    input_config: Optional[Union[_Config, Path, dict, str]] = None,
+    input_config: Optional[Union[Config, Path, dict, str]] = None,
     input_format: Optional[str] = None,
-    update_config: Optional[Union[_Config, Path, dict, str]] = None,
+    update_config: Optional[Union[Config, Path, dict, str]] = None,
     update_format: Optional[str] = None,
     output_file: Optional[Union[Path, str]] = None,
     output_format: Optional[str] = None,
@@ -143,9 +145,9 @@ def realize(
 
 
 def realize_to_dict(  # pylint: disable=unused-argument
-    input_config: Optional[Union[dict, _Config, Path, str]] = None,
+    input_config: Optional[Union[dict, Config, Path, str]] = None,
     input_format: Optional[str] = None,
-    update_config: Optional[Union[dict, _Config, Path, str]] = None,
+    update_config: Optional[Union[dict, Config, Path, str]] = None,
     update_format: Optional[str] = None,
     key_path: Optional[list[YAMLKey]] = None,
     values_needed: bool = False,
@@ -163,25 +165,32 @@ def realize_to_dict(  # pylint: disable=unused-argument
 
 def validate(
     schema_file: Union[Path, str],
-    config: Optional[Union[dict, YAMLConfig, Path, str]] = None,
+    config_data: Optional[ConfigDataT] = None,
+    config_path: Optional[ConfigPathT] = None,
     stdin_ok: bool = False,
 ) -> bool:
     """
     Check whether the specified config conforms to the specified JSON Schema spec.
 
     If no config is specified, ``stdin`` is read and will be parsed as YAML and then validated. A
-    ``dict`` or a YAMLConfig instance may also be provided for validation.
+    ``dict`` or a ``YAMLConfig`` instance may also be provided for validation.
 
     :param schema_file: The JSON Schema file to use for validation.
-    :param config: The config to validate.
+    :param config_data: A config to validate.
+    :param config_path: A path to a file containing a config to validate.
     :param stdin_ok: OK to read from ``stdin``?
+    :raises: TypeError if config_* arguments are not set appropriately.
     :return: ``True`` if the YAML file conforms to the schema, ``False`` otherwise.
     """
+    _validate_check_config(config_data, config_path)
+    if config_data is None:
+        config_path = _ensure_data_source(_str2path(config_path), stdin_ok)
     try:
         _validate_external(
             schema_file=_str2path(schema_file),
             desc="config",
-            config=_ensure_data_source(_str2path(config), stdin_ok),
+            config_data=config_data,
+            config_path=config_path,
         )
     except UWConfigError:
         return False
@@ -252,6 +261,7 @@ def validate(
 ).strip()
 
 __all__ = [
+    "Config",
     "FieldTableConfig",
     "INIConfig",
     "NMLConfig",

src/uwtools/cli.py

@@ -252,7 +252,7 @@ def _dispatch_config_validate(args: Args) -> bool:
     """
     return uwtools.api.config.validate(
         schema_file=args[STR.schemafile],
-        config=args[STR.infile],
+        config_path=args[STR.infile],
         stdin_ok=True,
     )
 

src/uwtools/config/formats/base.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import difflib
 import os
 import re
@@ -23,14 +25,17 @@ class Config(ABC, UserDict):
     several configuration-file formats.
     """
 
-    def __init__(self, config: Optional[Union[dict, str, Path]] = None) -> None:
+    def __init__(self, config: Optional[Union[dict, str, Config, Path]] = None) -> None:
         """
         :param config: Config file to load (None => read from stdin), or initial dict.
         """
         super().__init__()
         if isinstance(config, dict):
             self._config_file = None
             self.update(config)
+        elif isinstance(config, Config):
+            self._config_file = config._config_file
+            self.update(config.data)
         else:
             self._config_file = str2path(config) if config else None
             self.data = self._load(self._config_file)
@@ -216,7 +221,7 @@ def config_file(self) -> Optional[Path]:
         """
         return self._config_file
 
-    def dereference(self, context: Optional[dict] = None) -> None:
+    def dereference(self, context: Optional[dict] = None) -> Config:
         """
         Render as much Jinja2 syntax as possible.
         """
@@ -234,6 +239,7 @@ def logstate(state: str) -> None:
                 break
             self.data = new
         logstate("final")
+        return self
 
     @abstractmethod
     def dump(self, path: Optional[Path]) -> None:

src/uwtools/config/validator.py

@@ -19,6 +19,10 @@
 
 # Public functions
 
+JSONValueT = Union[bool, dict, float, int, list, str]
+ConfigDataT = Union[JSONValueT, YAMLConfig]
+ConfigPathT = Union[str, Path]
+
 
 def bundle(schema: dict, keys: Optional[list] = None) -> dict:
     """
@@ -57,7 +61,7 @@ def internal_schema_file(schema_name: str) -> Path:
     return resource_path("jsonschema") / f"{schema_name}.jsonschema"
 
 
-def validate(schema: dict, desc: str, config: dict) -> bool:
+def validate(schema: dict, desc: str, config: JSONValueT) -> bool:
     """
     Report any errors arising from validation of the given config against the given JSON Schema.
 
@@ -77,56 +81,79 @@ def validate(schema: dict, desc: str, config: dict) -> bool:
     return not bool(errors)
 
 
+def validate_check_config(
+    config_data: Optional[ConfigDataT] = None, config_path: Optional[ConfigPathT] = None
+) -> None:
+    """
+    Enforce mutual exclusivity of config_* arguments.
+
+    :param config_data: A config to validate.
+    :param config_path: A path to a file containing a config to validate.
+    :raises: TypeError if both config_* arguments specified.
+    """
+    if config_data is not None and config_path is not None:
+        raise TypeError("Specify at most one of config_data, config_path")
+
+
 def validate_internal(
-    schema_name: str, desc: str, config: Optional[Union[dict, YAMLConfig, Path]] = None
+    schema_name: str,
+    desc: str,
+    config_data: Optional[ConfigDataT] = None,
+    config_path: Optional[ConfigPathT] = None,
 ) -> None:
     """
     Validate a config against a uwtools-internal schema.
 
     :param schema_name: Name of uwtools schema to validate the config against.
     :param desc: A description of the config being validated, for logging.
-    :param config: The config to validate.
-    :raises: UWConfigError if config fails validation.
+    :param config_data: A config to validate.
+    :param config_path: A path to a file containing a config to validate.
+    :raises: TypeError if config_* arguments are not set appropriately.
     """
+    validate_check_config(config_data, config_path)
     log.info("Validating config against internal schema: %s", schema_name)
-    validate_external(config=config, schema_file=internal_schema_file(schema_name), desc=desc)
+    validate_external(
+        config_data=config_data,
+        config_path=config_path,
+        schema_file=internal_schema_file(schema_name),
+        desc=desc,
+    )
 
 
 def validate_external(
-    schema_file: Path, desc: str, config: Optional[Union[dict, YAMLConfig, Path]] = None
+    schema_file: Path,
+    desc: str,
+    config_data: Optional[ConfigDataT] = None,
+    config_path: Optional[ConfigPathT] = None,
 ) -> None:
     """
     Validate a YAML config against the JSON Schema in the given schema file.
 
     :param schema_file: The JSON Schema file to use for validation.
     :param desc: A description of the config being validated, for logging.
-    :param config: The config to validate.
-    :raises: UWConfigError if config fails validation.
-    """
+    :param config_data: A config to validate.
+    :param config_path: A path to a file containing a config to validate.
+    :raises: TypeError if config_* arguments are not set appropriately.
+    """
+    validate_check_config(config_data, config_path)
+    config: JSONValueT
+    if config_data is None:
+        config = YAMLConfig(config_path).dereference().data
+    elif isinstance(config_data, YAMLConfig):
+        config = config_data.data
+    else:
+        config = config_data
     if not str(schema_file).startswith(str(resource_path())):
         log.debug("Using schema file: %s", schema_file)
     with open(schema_file, "r", encoding="utf-8") as f:
         schema = json.load(f)
-    cfgobj = _prep_config(config)
-    if not validate(schema=schema, desc=desc, config=cfgobj.data):
+    if not validate(schema=schema, desc=desc, config=config):
         raise UWConfigError("YAML validation errors")
 
 
 # Private functions
 
 
-def _prep_config(config: Union[dict, YAMLConfig, Optional[Path]]) -> YAMLConfig:
-    """
-    Ensure a dereferenced YAMLConfig object for various input types.
-
-    :param config: The config to validate.
-    :return: A dereferenced YAMLConfig object based on the input config.
-    """
-    cfgobj = config if isinstance(config, YAMLConfig) else YAMLConfig(config)
-    cfgobj.dereference()
-    return cfgobj
-
-
 @cache
 def _registry() -> Registry:
     """
@@ -143,7 +170,7 @@ def retrieve(uri: str) -> Resource:
     return Registry(retrieve=retrieve)  # type: ignore
 
 
-def _validation_errors(config: Union[dict, list], schema: dict) -> list[ValidationError]:
+def _validation_errors(config: JSONValueT, schema: dict) -> list[ValidationError]:
     """
     Identify schema-validation errors.
 

src/uwtools/drivers/driver.py

@@ -46,7 +46,7 @@ def __init__(
         self,
         cycle: Optional[datetime] = None,
         leadtime: Optional[timedelta] = None,
-        config: Optional[Union[dict, str, YAMLConfig, Path]] = None,
+        config: Optional[Union[dict, str, Path, YAMLConfig]] = None,
         dry_run: bool = False,
         key_path: Optional[list[YAMLKey]] = None,
         schema_file: Optional[Path] = None,
@@ -233,7 +233,7 @@ def _validate(self) -> None:
         :raises: UWConfigError if config fails validation.
         """
         kwargs: dict = {
-            "config": self._config_intermediate,
+            "config_data": self._config_intermediate,
             "desc": "%s config" % self.driver_name(),
         }
         if self.schema_file:
@@ -536,7 +536,9 @@ def _validate(self) -> None:
         """
         Assets._validate(self)
         validate_internal(
-            schema_name=STR.platform, desc="platform config", config=self._config_intermediate
+            schema_name=STR.platform,
+            desc="platform config",
+            config_data=self._config_intermediate,
         )
 
     def _write_runscript(self, path: Path, envvars: Optional[dict[str, str]] = None) -> None:

src/uwtools/fs.py

@@ -116,7 +116,15 @@ def _validate(self) -> None:
 
         :raises: UWConfigError if config fails validation.
         """
-        validate_internal(schema_name=self._schema, desc="fs config", config=self._config)
+        config_data, config_path = (
+            (self._config, None) if isinstance(self._config, dict) else (None, self._config)
+        )
+        validate_internal(
+            schema_name=self._schema,
+            desc="fs config",
+            config_data=config_data,
+            config_path=config_path,
+        )
 
 
 class FileStager(Stager):

src/uwtools/resources/info.json

@@ -1,4 +1,4 @@
 {
   "buildnum": "0",
-  "version": "2.5.0"
+  "version": "2.6.0"
 }

src/uwtools/rocoto.py

@@ -350,15 +350,20 @@ def _add_workflow_tasks(self, e: _Element, config: dict) -> None:
             tag, name = self._tag_name(key)
             {STR.metatask: self._add_metatask, STR.task: self._add_task}[tag](e, subconfig, name)
 
-    def _config_validate(self, config: Union[dict, YAMLConfig, Optional[Path]]) -> None:
+    def _config_validate(self, config: Optional[Union[dict, Path, YAMLConfig]] = None) -> None:
         """
         Validate the given YAML config.
 
         :param config: YAMLConfig object or path to YAML file (None => read stdin).
         :raises: UWConfigError if config fails validation.
         """
-        schema_file = resource_path("jsonschema/rocoto.jsonschema")
-        validate_yaml(schema_file=schema_file, desc="Rocoto config", config=config)
+        config_data, config_path = (None, config) if isinstance(config, Path) else (config, None)
+        validate_yaml(
+            schema_file=resource_path("jsonschema/rocoto.jsonschema"),
+            desc="Rocoto config",
+            config_data=config_data,
+            config_path=config_path,
+        )
 
     @property
     def _doctype(self) -> Optional[str]: