omnibus-2024-04-15 (ufs-community#460)

docs/sections/user_guide/cli/tools/file.rst

@@ -107,7 +107,7 @@ Here, ``foo`` and ``bar`` are copies of their respective source files.
 ``link``
 --------
 
-The ``link`` action stages files in a target directory by linking files. Any ``KEY`` positional arguments are used to navigate, in the order given, from the top of the config to the :ref:`file block <files_yaml>`.
+The ``link`` action stages files in a target directory by linking files, directories, or other symbolic links. Any ``KEY`` positional arguments are used to navigate, in the order given, from the top of the config to the :ref:`file block <files_yaml>`.
 
 .. code-block:: text
 

recipe/meta.json

@@ -4,7 +4,7 @@
   "name": "uwtools",
   "packages": {
     "dev": [
-      "black =24.3.*",
+      "black =24.4.*",
       "coverage =7.4.*",
       "docformatter =1.7.*",
       "f90nml =1.4.*",

recipe/meta.yaml

@@ -22,7 +22,7 @@ requirements:
     - pyyaml 6.0.*
 test:
   requires:
-    - black 24.3.*
+    - black 24.4.*
     - coverage 7.4.*
     - docformatter 1.7.*
     - isort 5.13.*

src/uwtools/api/chgres_cube.py

@@ -4,21 +4,23 @@
 
 import datetime as dt
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 import iotaa as _iotaa
 
 import uwtools.drivers.support as _support
 from uwtools.drivers.chgres_cube import ChgresCube as _ChgresCube
+from uwtools.utils.api import ensure_data_source as _ensure_data_source
 
 
 def execute(
     task: str,
     cycle: dt.datetime,
-    config: Optional[Path] = None,
+    config: Optional[Union[Path, str]] = None,
     batch: bool = False,
     dry_run: bool = False,
-    graph_file: Optional[Path] = None,
+    graph_file: Optional[Union[Path, str]] = None,
+    stdin_ok: bool = False,
 ) -> bool:
     """
     Execute a ``chgres_cube`` task.
@@ -32,9 +34,12 @@ def execute(
     :param batch: Submit run to the batch system?
     :param dry_run: Do not run the executable, just report what would have been done.
     :param graph_file: Write Graphviz DOT output here.
+    :param stdin_ok: OK to read from stdin?
     :return: ``True`` if task completes without raising an exception.
     """
-    obj = _ChgresCube(config=config, cycle=cycle, batch=batch, dry_run=dry_run)
+    obj = _ChgresCube(
+        cycle=cycle, config=_ensure_data_source(config, stdin_ok), batch=batch, dry_run=dry_run
+    )
     getattr(obj, task)()
     if graph_file:
         with open(graph_file, "w", encoding="utf-8") as f:

src/uwtools/api/config.py

@@ -15,115 +15,142 @@
 from uwtools.config.tools import compare_configs as _compare
 from uwtools.config.tools import realize_config as _realize
 from uwtools.config.validator import validate_yaml as _validate_yaml
+from uwtools.utils.api import ensure_data_source as _ensure_data_source
+from uwtools.utils.api import str2path as _str2path
 from uwtools.utils.file import FORMAT as _FORMAT
 
 # Public
 
 
 def compare(
-    config_1_path: Path,
-    config_2_path: Path,
+    config_1_path: Union[Path, str],
+    config_2_path: Union[Path, str],
     config_1_format: Optional[str] = None,
     config_2_format: Optional[str] = None,
 ) -> bool:
     """
     NB: This docstring is dynamically replaced: See compare.__doc__ definition below.
     """
     return _compare(
-        config_1_path=config_1_path,
-        config_2_path=config_2_path,
+        config_1_path=Path(config_1_path),
+        config_2_path=Path(config_2_path),
         config_1_format=config_1_format,
         config_2_format=config_2_format,
     )
 
 
-def get_fieldtable_config(config: Union[dict, Optional[Path]] = None) -> _FieldTableConfig:
+def get_fieldtable_config(
+    config: Union[dict, Optional[Union[Path, str]]] = None, stdin_ok=False
+) -> _FieldTableConfig:
     """
     Get a ``FieldTableConfig`` object.
 
     :param config: FieldTable file to load (``None`` or unspecified => read ``stdin``), or initial
         ``dict``
+    :param stdin_ok: OK to read from stdin?
     :return: An initialized ``FieldTableConfig`` object
     """
-    return _FieldTableConfig(config=config)
+    return _FieldTableConfig(config=_ensure_data_source(config, stdin_ok))
 
 
-def get_ini_config(config: Union[dict, Optional[Path]] = None) -> _INIConfig:
+def get_ini_config(
+    config: Union[dict, Optional[Union[Path, str]]] = None,
+    stdin_ok: bool = False,
+) -> _INIConfig:
     """
     Get an ``INIConfig`` object.
 
     :param config: INI file to load (``None`` or unspecified => read ``stdin``), or initial ``dict``
+    :param stdin_ok: OK to read from stdin?
     :return: An initialized ``INIConfig`` object
     """
-    return _INIConfig(config=config)
+    return _INIConfig(config=_ensure_data_source(config, stdin_ok))
 
 
-def get_nml_config(config: Union[dict, Optional[Path]] = None) -> _NMLConfig:
+def get_nml_config(
+    config: Union[dict, Optional[Union[Path, str]]] = None,
+    stdin_ok: bool = False,
+) -> _NMLConfig:
     """
     Get an ``NMLConfig`` object.
 
     :param config: Fortran namelist file to load (``None`` or unspecified => read ``stdin``), or
         initial ``dict``
+    :param stdin_ok: OK to read from stdin?
     :return: An initialized ``NMLConfig`` object
     """
-    return _NMLConfig(config=config)
+    return _NMLConfig(config=_ensure_data_source(config, stdin_ok))
 
 
-def get_sh_config(config: Union[dict, Optional[Path]] = None) -> _SHConfig:
+def get_sh_config(
+    config: Union[dict, Optional[Union[Path, str]]] = None,
+    stdin_ok: bool = False,
+) -> _SHConfig:
     """
     Get an ``SHConfig`` object.
 
     :param config: File of shell 'key=value' pairs to load (``None`` or unspecified => read
         ``stdin``), or initial ``dict``
+    :param stdin_ok: OK to read from stdin?
     :return: An initialized ``SHConfig`` object
     """
-    return _SHConfig(config=config)
+    return _SHConfig(config=_ensure_data_source(config, stdin_ok))
 
 
-def get_yaml_config(config: Union[dict, Optional[Path]] = None) -> _YAMLConfig:
+def get_yaml_config(
+    config: Union[dict, Optional[Union[Path, str]]] = None,
+    stdin_ok: bool = False,
+) -> _YAMLConfig:
     """
     Get a ``YAMLConfig`` object.
 
     :param config: YAML file to load (``None`` or unspecified => read ``stdin``), or initial
         ``dict``
+    :param stdin_ok: OK to read from stdin?
     :return: An initialized ``YAMLConfig`` object
     """
-    return _YAMLConfig(config=config)
+    return _YAMLConfig(config=_ensure_data_source(config, stdin_ok))
 
 
 def realize(
-    input_config: Union[dict, _Config, Optional[Path]] = None,
+    input_config: Optional[Union[dict, _Config, Path, str]] = None,
     input_format: Optional[str] = None,
     output_block: Optional[List[Union[str, int]]] = None,
-    output_file: Optional[Path] = None,
+    output_file: Optional[Union[Path, str]] = None,
     output_format: Optional[str] = None,
-    supplemental_configs: Optional[List[Union[dict, _Config, Path]]] = None,
+    supplemental_configs: Optional[List[Union[dict, _Config, Path, str]]] = None,
     values_needed: bool = False,
     total: bool = False,
     dry_run: bool = False,
+    stdin_ok: bool = False,
 ) -> None:
     """
     NB: This docstring is dynamically replaced: See realize.__doc__ definition below.
     """
+    input_config = (
+        _YAMLConfig(config=input_config) if isinstance(input_config, dict) else input_config
+    )
+    scs = [_str2path(x) for x in supplemental_configs] if supplemental_configs else None
     _realize(
-        input_config=_ensure_config_arg_type(input_config),
+        input_config=_ensure_data_source(input_config, stdin_ok),
         input_format=input_format,
         output_block=output_block,
-        output_file=output_file,
+        output_file=_str2path(output_file),
         output_format=output_format,
-        supplemental_configs=supplemental_configs,
+        supplemental_configs=scs,
         values_needed=values_needed,
         total=total,
         dry_run=dry_run,
     )
 
 
 def realize_to_dict(  # pylint: disable=unused-argument
-    input_config: Union[dict, _Config, Optional[Path]] = None,
+    input_config: Optional[Union[dict, _Config, Path, str]] = None,
     input_format: Optional[str] = None,
-    supplemental_configs: Optional[List[Union[dict, _Config, Path]]] = None,
+    supplemental_configs: Optional[List[Union[dict, _Config, Path, str]]] = None,
     values_needed: bool = False,
     dry_run: bool = False,
+    stdin_ok: bool = False,
 ) -> dict:
     """
     Realize a config to a ``dict``, based on an input config and optional supplemental configs.
@@ -134,7 +161,9 @@ def realize_to_dict(  # pylint: disable=unused-argument
 
 
 def validate(
-    schema_file: Path, config: Optional[Union[dict, _YAMLConfig, Optional[Path]]] = None
+    schema_file: Union[Path, str],
+    config: Optional[Union[dict, _YAMLConfig, Path, str]] = None,
+    stdin_ok: bool = False,
 ) -> bool:
     """
     Check whether the specified config conforms to the specified JSON Schema spec.
@@ -144,25 +173,12 @@ def validate(
 
     :param schema_file: The JSON Schema file to use for validation
     :param config: The config to validate
+    :param stdin_ok: OK to read from stdin?
     :return: ``True`` if the YAML file conforms to the schema, ``False`` otherwise
     """
-    return _validate_yaml(schema_file=schema_file, config=config)
-
-
-# Private
-
-
-def _ensure_config_arg_type(
-    config: Union[dict, _Config, Optional[Path]]
-) -> Union[_Config, Optional[Path]]:
-    """
-    Encapsulate a ``dict`` in a ``Config``; return a ``Config`` or path argument as-is.
-
-    :param config: A config as a ``dict``, ``Config``, or ``Path``
-    """
-    if isinstance(config, dict):
-        return _YAMLConfig(config=config)
-    return config
+    return _validate_yaml(
+        schema_file=_ensure_data_source(schema_file, stdin_ok), config=_str2path(config)
+    )
 
 
 # Import-time code
@@ -219,6 +235,7 @@ def _ensure_config_arg_type(
 :param values_needed: Report complete, missing, and template values
 :param total: Require rendering of all Jinja2 variables/expressions
 :param dry_run: Log output instead of writing to output
+:param stdin_ok: OK to read from stdin?
 :raises: UWConfigRealizeError if ``total`` is ``True`` and any Jinja2 variable/expression was not rendered
 """.format(
     extensions=", ".join(_FORMAT.extensions())

src/uwtools/api/file.py

@@ -7,13 +7,15 @@
 
 from uwtools.file import FileCopier as _FileCopier
 from uwtools.file import FileLinker as _FileLinker
+from uwtools.utils.api import ensure_data_source as _ensure_data_source
 
 
 def copy(
-    target_dir: Path,
-    config: Optional[Union[dict, Path]] = None,
+    target_dir: Union[Path, str],
+    config: Optional[Union[dict, Path, str]] = None,
     keys: Optional[List[str]] = None,
     dry_run: bool = False,
+    stdin_ok: bool = False,
 ) -> bool:
     """
     Copy files.
@@ -22,17 +24,24 @@ def copy(
     :param config: YAML-file path, or ``dict`` (read ``stdin`` if missing or ``None``).
     :param keys: YAML keys leading to file dst/src block
     :param dry_run: Do not copy files
+    :param stdin_ok: OK to read from stdin?
     :return: ``True`` if no exception is raised
     """
-    _FileCopier(target_dir=target_dir, config=config, keys=keys, dry_run=dry_run).go()
+    _FileCopier(
+        target_dir=Path(target_dir),
+        config=_ensure_data_source(config, stdin_ok),
+        keys=keys,
+        dry_run=dry_run,
+    ).go()
     return True
 
 
 def link(
-    target_dir: Path,
-    config: Optional[Union[dict, Path]] = None,
+    target_dir: Union[Path, str],
+    config: Optional[Union[dict, Path, str]] = None,
     keys: Optional[List[str]] = None,
     dry_run: bool = False,
+    stdin_ok: bool = False,
 ) -> bool:
     """
     Link files.
@@ -41,7 +50,13 @@ def link(
     :param config: YAML-file path, or ``dict`` (read ``stdin`` if missing or ``None``).
     :param keys: YAML keys leading to file dst/src block
     :param dry_run: Do not link files
+    :param stdin_ok: OK to read from stdin?
     :return: ``True`` if no exception is raised
     """
-    _FileLinker(target_dir=target_dir, config=config, keys=keys, dry_run=dry_run).go()
+    _FileLinker(
+        target_dir=Path(target_dir),
+        config=_ensure_data_source(config, stdin_ok),
+        keys=keys,
+        dry_run=dry_run,
+    ).go()
     return True

src/uwtools/api/fv3.py

@@ -4,19 +4,21 @@
 
 import datetime as dt
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 import uwtools.drivers.support as _support
 from uwtools.drivers.fv3 import FV3 as _FV3
+from uwtools.utils.api import ensure_data_source as _ensure_data_source
 
 
 def execute(
     task: str,
     cycle: dt.datetime,
-    config: Optional[Path] = None,
+    config: Optional[Union[Path, str]] = None,
     batch: bool = False,
     dry_run: bool = False,
     graph_file: Optional[Path] = None,
+    stdin_ok: bool = False,
 ) -> bool:
     """
     Execute an FV3 task.
@@ -30,9 +32,12 @@ def execute(
     :param batch: Submit run to the batch system?
     :param dry_run: Do not run forecast, just report what would have been done.
     :param graph_file: Write Graphviz DOT output here.
+    :param stdin_ok: OK to read from stdin?
     :return: ``True`` if task completes without raising an exception.
     """
-    obj = _FV3(config=config, cycle=cycle, batch=batch, dry_run=dry_run)
+    obj = _FV3(
+        cycle=cycle, config=_ensure_data_source(config, stdin_ok), batch=batch, dry_run=dry_run
+    )
     getattr(obj, task)()
     if graph_file:
         with open(graph_file, "w", encoding="utf-8") as f: