Add typed task input forms

2026-08-08 18:39:09 +00:00
parent 56d6194324
commit 0549ed1b0d
8 changed files with 521 additions and 16 deletions
@@ -3,8 +3,15 @@
 from __future__ import annotations

 import ast
+import json
+import re
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Any
+
+
+FIELD_NAME = re.compile(r"[a-z][a-z0-9_]*\Z")
+FIELD_TYPES = frozenset({"text", "integer", "date", "datetime", "choice", "multi_choice"})


 def display_name(value: str) -> str:
@@ -19,6 +26,27 @@ class Task:
    path: Path
    display_name: str
    description: str | None
+    inputs: tuple["InputField", ...]
+
+
+@dataclass(frozen=True)
+class InputOption:
+    value: str
+    label: str
+
+
+@dataclass(frozen=True)
+class InputField:
+    name: str
+    label: str
+    type: str
+    required: bool
+    default: str | int | tuple[str, ...] | None
+    minimum: int | None
+    maximum: int | None
+    step: int | None
+    pattern: str | None
+    options: tuple[InputOption, ...]


@dataclass(frozen=True)
@@ -69,15 +97,22 @@ class TaskCatalog:
        return None

    def _tasks_in_group(self, group_path: Path) -> list[Task]:
-        tasks = []
-        for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name):
-            if (
+        task_paths = [
+            path
+            for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name)
+            if not (
                path.name.startswith((".", "_"))
                or path.is_symlink()
                or not path.is_file()
                or path.suffix != ".py"
-            ):
-                continue
+            )
+        ]
+        inputs_by_filename = load_input_manifest(
+            group_path,
+            {path.name for path in task_paths},
+        )
+        tasks = []
+        for path in task_paths:
            tasks.append(
                Task(
                    id=f"{group_path.name}/{path.name}",
@@ -86,6 +121,7 @@ class TaskCatalog:
                    path=path,
                    display_name=display_name(path.stem),
                    description=read_description(path),
+                    inputs=inputs_by_filename.get(path.name, ()),
                )
            )
        return tasks
@@ -105,3 +141,214 @@ def read_description(path: Path) -> str | None:
        (line.strip() for line in docstring.splitlines() if line.strip()),
        None,
    )
+
+
+def load_input_manifest(
+    group_path: Path,
+    task_filenames: set[str],
+) -> dict[str, tuple[InputField, ...]]:
+    """Read the optional data-only input declaration for one task group."""
+
+    manifest_path = group_path / "task-inputs.json"
+    if not manifest_path.exists():
+        return {}
+    if manifest_path.is_symlink():
+        raise ValueError(f"Input manifest cannot be a symlink: {manifest_path}")
+    try:
+        data = json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise ValueError(f"Could not read input manifest {manifest_path}: {error}") from error
+    if not isinstance(data, dict) or set(data) != {"version", "tasks"}:
+        raise ValueError(
+            f"Input manifest {manifest_path} must contain only version and tasks."
+        )
+    if data["version"] != 1 or not isinstance(data["tasks"], dict):
+        raise ValueError(f"Input manifest {manifest_path} has an unsupported structure.")
+
+    unknown_tasks = set(data["tasks"]) - task_filenames
+    if unknown_tasks:
+        names = ", ".join(sorted(unknown_tasks))
+        raise ValueError(f"Input manifest {manifest_path} names unknown tasks: {names}.")
+
+    return {
+        filename: parse_task_inputs(manifest_path, filename, declaration)
+        for filename, declaration in data["tasks"].items()
+    }
+
+
+def parse_task_inputs(
+    manifest_path: Path,
+    filename: str,
+    declaration: Any,
+) -> tuple[InputField, ...]:
+    if not isinstance(declaration, dict) or set(declaration) != {"inputs"}:
+        raise ValueError(f"{manifest_path} task {filename} must contain only inputs.")
+    raw_inputs = declaration["inputs"]
+    if not isinstance(raw_inputs, list):
+        raise ValueError(f"{manifest_path} task {filename} inputs must be a list.")
+    fields = tuple(parse_input_field(manifest_path, filename, raw) for raw in raw_inputs)
+    names = [field.name for field in fields]
+    if len(names) != len(set(names)):
+        raise ValueError(f"{manifest_path} task {filename} has duplicate input names.")
+    return fields
+
+
+def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputField:
+    if not isinstance(raw, dict):
+        raise ValueError(f"{manifest_path} task {filename} has a non-object input field.")
+    allowed = {
+        "name", "label", "type", "required", "default", "minimum", "maximum",
+        "step", "pattern", "options",
+    }
+    unknown = set(raw) - allowed
+    if unknown:
+        raise ValueError(
+            f"{manifest_path} task {filename} input has unknown keys: {', '.join(sorted(unknown))}."
+        )
+    name = raw.get("name")
+    label = raw.get("label")
+    field_type = raw.get("type")
+    if not isinstance(name, str) or not FIELD_NAME.fullmatch(name):
+        raise ValueError(f"{manifest_path} task {filename} has an invalid input name.")
+    if not isinstance(label, str) or not label.strip():
+        raise ValueError(f"{manifest_path} task {filename} input {name} needs a label.")
+    if field_type not in FIELD_TYPES:
+        raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid type.")
+    required = raw.get("required", False)
+    if not isinstance(required, bool):
+        raise ValueError(f"{manifest_path} task {filename} input {name} required must be true or false.")
+
+    numeric_keys = ("minimum", "maximum", "step")
+    numeric_values = {key: raw.get(key) for key in numeric_keys}
+    if field_type == "integer":
+        for key, value in numeric_values.items():
+            if value is not None and (not isinstance(value, int) or isinstance(value, bool)):
+                raise ValueError(f"{manifest_path} task {filename} input {name} {key} must be an integer.")
+        if numeric_values["minimum"] is not None and numeric_values["maximum"] is not None and numeric_values["minimum"] > numeric_values["maximum"]:
+            raise ValueError(f"{manifest_path} task {filename} input {name} minimum exceeds maximum.")
+        if numeric_values["step"] is not None and numeric_values["step"] <= 0:
+            raise ValueError(f"{manifest_path} task {filename} input {name} step must be positive.")
+    elif any(value is not None for value in numeric_values.values()):
+        raise ValueError(f"{manifest_path} task {filename} input {name} only integer fields accept numeric limits.")
+
+    pattern = raw.get("pattern")
+    if pattern is not None:
+        if field_type != "text" or not isinstance(pattern, str):
+            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern.")
+        try:
+            re.compile(pattern)
+        except re.error as error:
+            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern: {error}.") from error
+
+    options = parse_options(manifest_path, filename, name, field_type, raw.get("options"))
+    default = parse_default(manifest_path, filename, name, field_type, options, raw.get("default"))
+    validate_default(
+        manifest_path,
+        filename,
+        name,
+        field_type,
+        default,
+        numeric_values["minimum"],
+        numeric_values["maximum"],
+        numeric_values["step"],
+        pattern,
+    )
+    return InputField(
+        name=name,
+        label=label,
+        type=field_type,
+        required=required,
+        default=default,
+        minimum=numeric_values["minimum"],
+        maximum=numeric_values["maximum"],
+        step=numeric_values["step"],
+        pattern=pattern,
+        options=options,
+    )
+
+
+def parse_options(
+    manifest_path: Path,
+    filename: str,
+    name: str,
+    field_type: str,
+    raw_options: Any,
+) -> tuple[InputOption, ...]:
+    if field_type not in {"choice", "multi_choice"}:
+        if raw_options is not None:
+            raise ValueError(f"{manifest_path} task {filename} input {name} only choice fields accept options.")
+        return ()
+    if not isinstance(raw_options, list) or not raw_options:
+        raise ValueError(f"{manifest_path} task {filename} input {name} needs non-empty options.")
+    options = []
+    for raw_option in raw_options:
+        if isinstance(raw_option, str) and raw_option:
+            options.append(InputOption(value=raw_option, label=raw_option))
+        elif (
+            isinstance(raw_option, dict)
+            and set(raw_option) == {"value", "label"}
+            and isinstance(raw_option["value"], str)
+            and raw_option["value"]
+            and isinstance(raw_option["label"], str)
+            and raw_option["label"].strip()
+        ):
+            options.append(InputOption(value=raw_option["value"], label=raw_option["label"]))
+        else:
+            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid option.")
+    if len({option.value for option in options}) != len(options):
+        raise ValueError(f"{manifest_path} task {filename} input {name} has duplicate option values.")
+    return tuple(options)
+
+
+def parse_default(
+    manifest_path: Path,
+    filename: str,
+    name: str,
+    field_type: str,
+    options: tuple[InputOption, ...],
+    default: Any,
+) -> str | int | tuple[str, ...] | None:
+    if default is None:
+        return None
+    if field_type == "integer":
+        if not isinstance(default, int) or isinstance(default, bool):
+            raise ValueError(f"{manifest_path} task {filename} input {name} default must be an integer.")
+        return default
+    if field_type == "multi_choice":
+        if not isinstance(default, list) or not all(isinstance(value, str) for value in default):
+            raise ValueError(f"{manifest_path} task {filename} input {name} default must be a list of options.")
+        values = tuple(default)
+        if len(values) != len(set(values)) or not set(values) <= {option.value for option in options}:
+            raise ValueError(f"{manifest_path} task {filename} input {name} default contains an unknown option.")
+        return values
+    if not isinstance(default, str):
+        raise ValueError(f"{manifest_path} task {filename} input {name} default must be text.")
+    if field_type == "choice" and default not in {option.value for option in options}:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default is not an option.")
+    return default
+
+
+def validate_default(
+    manifest_path: Path,
+    filename: str,
+    name: str,
+    field_type: str,
+    default: str | int | tuple[str, ...] | None,
+    minimum: int | None,
+    maximum: int | None,
+    step: int | None,
+    pattern: str | None,
+) -> None:
+    if default is None or field_type not in {"integer", "text"}:
+        return
+    if field_type == "text":
+        if pattern and not re.fullmatch(pattern, default):
+            raise ValueError(f"{manifest_path} task {filename} input {name} default does not match its pattern.")
+        return
+    assert isinstance(default, int)
+    if minimum is not None and default < minimum:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default is below minimum.")
+    if maximum is not None and default > maximum:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default is above maximum.")
+    if step is not None and (default - (minimum or 0)) % step:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default does not match step.")