Normalize programmatic build_dag config defaults (#767)

Author: tobiasraabe

src/_pytask/build.py

@@ -5,7 +5,6 @@
 import json
 import sys
 from contextlib import suppress
-from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Literal
@@ -16,8 +15,7 @@
 from _pytask.capture_utils import CaptureMethod
 from _pytask.capture_utils import ShowCapture
 from _pytask.click import ColoredCommand
-from _pytask.config_utils import find_project_root_and_config
-from _pytask.config_utils import read_config
+from _pytask.config_utils import normalize_programmatic_config
 from _pytask.console import console
 from _pytask.dag import create_dag
 from _pytask.exceptions import CollectionError
@@ -30,13 +28,12 @@
 from _pytask.pluginmanager import hookimpl
 from _pytask.pluginmanager import storage
 from _pytask.session import Session
-from _pytask.shared import parse_paths
-from _pytask.shared import to_list
 from _pytask.traceback import Traceback
 
 if TYPE_CHECKING:
     from collections.abc import Callable
     from collections.abc import Iterable
+    from pathlib import Path
     from typing import NoReturn
 
     from _pytask.node_protocols import PTask
@@ -66,7 +63,7 @@ def pytask_unconfigure(session: Session) -> None:
     path.write_text(json.dumps(HashPathCache._cache))
 
 
-def build(  # noqa: C901, PLR0912, PLR0913, PLR0915
+def build(  # noqa: PLR0913
     *,
     capture: Literal["fd", "no", "sys", "tee-sys"] | CaptureMethod = CaptureMethod.FD,
     check_casing_of_paths: bool = True,
@@ -225,47 +222,14 @@ def build(  # noqa: C901, PLR0912, PLR0913, PLR0915
 
         # If someone called the programmatic interface, we need to do some parsing.
         if "command" not in raw_config:
-            raw_config["command"] = "build"
             # Add defaults from cli.
             from _pytask.cli import DEFAULTS_FROM_CLI  # noqa: PLC0415
 
-            raw_config = {**DEFAULTS_FROM_CLI, **raw_config}
-
-            paths_value = raw_config["paths"]
-            # Convert tuple to list since parse_paths expects Path | list[Path]
-            if isinstance(paths_value, tuple):
-                paths_value = list(paths_value)
-            if not isinstance(paths_value, (Path, list)):
-                msg = f"paths must be Path or list, got {type(paths_value)}"
-                raise TypeError(msg)  # noqa: TRY301
-            # Cast is justified - we validated at runtime
-            raw_config["paths"] = parse_paths(cast("Path | list[Path]", paths_value))
-
-            if raw_config["config"] is not None:
-                config_value = raw_config["config"]
-                if not isinstance(config_value, (str, Path)):
-                    msg = f"config must be str or Path, got {type(config_value)}"
-                    raise TypeError(msg)  # noqa: TRY301
-                raw_config["config"] = Path(config_value).resolve()
-                raw_config["root"] = raw_config["config"].parent
-            else:
-                (
-                    raw_config["root"],
-                    raw_config["config"],
-                ) = find_project_root_and_config(raw_config["paths"])
-
-            if raw_config["config"] is not None:
-                config_from_file = read_config(raw_config["config"])
-
-                if "paths" in config_from_file:
-                    paths = config_from_file["paths"]
-                    paths = [
-                        raw_config["config"].parent.joinpath(path).resolve()
-                        for path in to_list(paths)
-                    ]
-                    config_from_file["paths"] = paths
-
-                raw_config = {**raw_config, **config_from_file}
+            raw_config = normalize_programmatic_config(
+                raw_config,
+                command="build",
+                defaults_from_cli=cast("dict[str, Any]", DEFAULTS_FROM_CLI),
+            )
 
         config_ = pm.hook.pytask_configure(pm=pm, raw_config=raw_config)
 

src/_pytask/config_utils.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import enum
 import os
 import sys
 from pathlib import Path
@@ -12,6 +13,7 @@
 import click
 
 from _pytask.shared import parse_paths
+from _pytask.shared import to_list
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -22,7 +24,12 @@
     import tomli as tomllib  # ty: ignore[unresolved-import]
 
 
-__all__ = ["find_project_root_and_config", "read_config", "set_defaults_from_config"]
+__all__ = [
+    "find_project_root_and_config",
+    "normalize_programmatic_config",
+    "read_config",
+    "set_defaults_from_config",
+]
 
 
 def set_defaults_from_config(
@@ -78,6 +85,73 @@ def set_defaults_from_config(
     return context.params["config"]
 
 
+def normalize_programmatic_config(
+    raw_config: dict[str, Any],
+    *,
+    command: str,
+    defaults_from_cli: dict[str, Any],
+) -> dict[str, Any]:
+    """Normalize programmatic raw_config inputs."""
+    raw_config = {**defaults_from_cli, **raw_config}
+    raw_config["command"] = command
+    raw_config["paths"] = _normalize_paths_value(raw_config["paths"])
+    _normalize_config_value(raw_config)
+
+    if raw_config["config"] is not None:
+        config_from_file = read_config(raw_config["config"])
+
+        if "paths" in config_from_file:
+            paths = config_from_file["paths"]
+            paths = [
+                raw_config["config"].parent.joinpath(path).resolve()
+                for path in to_list(paths)
+            ]
+            config_from_file["paths"] = paths
+
+        raw_config = {**raw_config, **config_from_file}
+
+    return raw_config
+
+
+def _normalize_paths_value(paths_value: Any) -> list[Path]:
+    """Normalize paths from programmatic inputs."""
+    if isinstance(paths_value, tuple):
+        paths_value = list(paths_value)
+    if not isinstance(paths_value, (Path, list)):
+        msg = f"paths must be Path or list, got {type(paths_value)}"
+        raise TypeError(msg)
+    # Cast is justified - we validated at runtime
+    return parse_paths(cast("Path | list[Path]", paths_value))
+
+
+def _normalize_config_value(raw_config: dict[str, Any]) -> None:
+    """Normalize config value from programmatic inputs."""
+    config_value = raw_config["config"]
+    if _is_click_sentinel(config_value):
+        config_value = None
+        raw_config["config"] = None
+    if config_value is not None:
+        if not isinstance(config_value, (str, Path)):
+            msg = f"config must be str or Path, got {type(config_value)}"
+            raise TypeError(msg)
+        raw_config["config"] = Path(config_value).resolve()
+        raw_config["root"] = raw_config["config"].parent
+    else:
+        raw_config["root"], raw_config["config"] = find_project_root_and_config(
+            raw_config["paths"]
+        )
+
+
+def _is_click_sentinel(value: Any) -> bool:
+    """Return True if value looks like Click's Sentinel.UNSET."""
+    return (
+        isinstance(value, enum.Enum)
+        and value.name == "UNSET"
+        and value.__class__.__name__ == "Sentinel"
+        and value.__class__.__module__ == "click._utils"
+    )
+
+
 def find_project_root_and_config(
     paths: Sequence[Path] | None,
 ) -> tuple[Path, Path | None]:

src/_pytask/dag_command.py

@@ -16,8 +16,7 @@
 from _pytask.click import EnumChoice
 from _pytask.compat import check_for_optional_program
 from _pytask.compat import import_optional_dependency
-from _pytask.config_utils import find_project_root_and_config
-from _pytask.config_utils import read_config
+from _pytask.config_utils import normalize_programmatic_config
 from _pytask.console import console
 from _pytask.dag import create_dag
 from _pytask.exceptions import CollectionError
@@ -28,9 +27,7 @@
 from _pytask.pluginmanager import hookimpl
 from _pytask.pluginmanager import storage
 from _pytask.session import Session
-from _pytask.shared import parse_paths
 from _pytask.shared import reduce_names_of_multiple_nodes
-from _pytask.shared import to_list
 from _pytask.traceback import Traceback
 
 
@@ -150,47 +147,14 @@ def build_dag(raw_config: dict[str, Any]) -> nx.DiGraph:
 
         # If someone called the programmatic interface, we need to do some parsing.
         if "command" not in raw_config:
-            raw_config["command"] = "dag"
             # Add defaults from cli.
             from _pytask.cli import DEFAULTS_FROM_CLI  # noqa: PLC0415
 
-            raw_config = {**DEFAULTS_FROM_CLI, **raw_config}  # ty: ignore[invalid-assignment]
-
-            paths_value = raw_config["paths"]
-            # Convert tuple to list since parse_paths expects Path | list[Path]
-            if isinstance(paths_value, tuple):
-                paths_value = list(paths_value)
-            if not isinstance(paths_value, (Path, list)):
-                msg = f"paths must be Path or list, got {type(paths_value)}"
-                raise TypeError(msg)  # noqa: TRY301
-            # Cast is justified - we validated at runtime
-            raw_config["paths"] = parse_paths(cast("Path | list[Path]", paths_value))
-
-            if raw_config["config"] is not None:
-                config_value = raw_config["config"]
-                if not isinstance(config_value, (str, Path)):
-                    msg = f"config must be str or Path, got {type(config_value)}"
-                    raise TypeError(msg)  # noqa: TRY301
-                raw_config["config"] = Path(config_value).resolve()
-                raw_config["root"] = raw_config["config"].parent
-            else:
-                (
-                    raw_config["root"],
-                    raw_config["config"],
-                ) = find_project_root_and_config(raw_config["paths"])
-
-            if raw_config["config"] is not None:
-                config_from_file = read_config(raw_config["config"])
-
-                if "paths" in config_from_file:
-                    paths = config_from_file["paths"]
-                    paths = [
-                        raw_config["config"].parent.joinpath(path).resolve()
-                        for path in to_list(paths)
-                    ]
-                    config_from_file["paths"] = paths
-
-                raw_config = {**raw_config, **config_from_file}
+            raw_config = normalize_programmatic_config(
+                raw_config,
+                command="dag",
+                defaults_from_cli=cast("dict[str, Any]", DEFAULTS_FROM_CLI),
+            )
 
         config = pm.hook.pytask_configure(pm=pm, raw_config=raw_config)