|
| 1 | +"""``graph_input()`` / ``graph_output()`` — declare graph I/O from the body. |
| 2 | +
|
| 3 | +A pipeline's inputs normally come from its ``In[T]`` parameters and its |
| 4 | +outputs from the return annotation. That covers Python-shaped graphs only: |
| 5 | +it cannot express a name that is not an identifier (``"Pipeline Creation |
| 6 | +Time"``), an exact Tangle type string, an input declared conditionally, or |
| 7 | +a per-input ``editor.position``. These two functions declare the same |
| 8 | +entries directly on the active trace, so a YAML pipeline that uses those |
| 9 | +shapes stays expressible in Python. |
| 10 | +
|
| 11 | +Entry key order follows the tracer's signature path and the corpus: |
| 12 | +``name, type, description, default, optional, annotations``. |
| 13 | +""" |
| 14 | +from __future__ import annotations |
| 15 | + |
| 16 | +from collections.abc import Mapping |
| 17 | +from typing import Any |
| 18 | + |
| 19 | +from tangle_cli.editor_layout import POSITION_ANNOTATION, position_annotation_value |
| 20 | +from tangle_cli.schema_validation import CALLER_ANNOTATION_POLICY, check_annotations |
| 21 | + |
| 22 | +from .errors import InvalidEditorLayoutError, InvalidGraphIoError |
| 23 | +from .graph import EdgeRef |
| 24 | +from .placeholders import GraphInputPlaceholder, TaskOutputProxy |
| 25 | + |
| 26 | +# Distinguishes "no default" from an explicit ``default=None``, which is a |
| 27 | +# value the schema does not accept and must be reported, not dropped. |
| 28 | +_UNSET: Any = object() |
| 29 | + |
| 30 | +__all__ = ["graph_input", "graph_output"] |
| 31 | + |
| 32 | + |
| 33 | +def graph_input( |
| 34 | + name: str, |
| 35 | + type: str | None = None, |
| 36 | + *, |
| 37 | + description: str | None = None, |
| 38 | + default: Any = _UNSET, |
| 39 | + optional: bool | None = None, |
| 40 | + annotations: Mapping[str, str] | None = None, |
| 41 | + position: tuple[float, float] | None = None, |
| 42 | + when: bool = True, |
| 43 | +) -> GraphInputPlaceholder | None: |
| 44 | + """Declare a graph input on the active pipeline and return its handle. |
| 45 | +
|
| 46 | + The handle is passed to task arguments exactly like an ``In[T]`` |
| 47 | + parameter. Supplying ``default`` implies ``optional: true`` unless |
| 48 | + ``optional=False`` is passed explicitly. |
| 49 | +
|
| 50 | + ``when=False`` declares nothing and returns ``None``, so a conditional |
| 51 | + input reads as one assignment instead of an ``if``/``else``. |
| 52 | +
|
| 53 | + Raises: |
| 54 | + InvalidGraphIoError: Outside a ``@pipeline`` body, on a duplicate |
| 55 | + name, or on an unusable field. Messages name the field, never |
| 56 | + the value. |
| 57 | + """ |
| 58 | + builder = _active_builder("graph_input") |
| 59 | + if not when: |
| 60 | + return None |
| 61 | + |
| 62 | + _check_name(name, "graph_input") |
| 63 | + if any(entry.get("name") == name for entry in builder.inputs): |
| 64 | + # Also catches an ``In[...]`` parameter: signature inputs are |
| 65 | + # appended before the body runs. |
| 66 | + raise InvalidGraphIoError( |
| 67 | + f"graph input {name!r} is already declared on pipeline " |
| 68 | + f"{builder.name!r}. Input names must be unique." |
| 69 | + ) |
| 70 | + |
| 71 | + entry: dict[str, Any] = {"name": name} |
| 72 | + if type is not None: |
| 73 | + entry["type"] = _check_str("type", type) |
| 74 | + if description is not None: |
| 75 | + entry["description"] = _check_str("description", description) |
| 76 | + if default is not _UNSET: |
| 77 | + # InputSpec.default is a string in the schema; a non-string default |
| 78 | + # would only fail later, against a schema path instead of this call. |
| 79 | + entry["default"] = _check_str("default", default, hint="pass str(value)") |
| 80 | + entry["optional"] = True if optional is None else _check_bool(optional) |
| 81 | + elif optional is not None: |
| 82 | + entry["optional"] = _check_bool(optional) |
| 83 | + merged = _layout_annotations(annotations, position) |
| 84 | + if merged: |
| 85 | + entry["annotations"] = merged |
| 86 | + |
| 87 | + builder.inputs.append(entry) |
| 88 | + return GraphInputPlaceholder(input_name=name) |
| 89 | + |
| 90 | + |
| 91 | +def graph_output( |
| 92 | + name: str, |
| 93 | + value: Any, |
| 94 | + type: str | None = None, |
| 95 | + *, |
| 96 | + description: str | None = None, |
| 97 | + annotations: Mapping[str, str] | None = None, |
| 98 | + position: tuple[float, float] | None = None, |
| 99 | + when: bool = True, |
| 100 | +) -> None: |
| 101 | + """Declare a graph output on the active pipeline, wired to ``value``. |
| 102 | +
|
| 103 | + ``value`` is a task output handle or a graph-input handle; a constant is |
| 104 | + refused, matching the ``Outputs`` return path. |
| 105 | +
|
| 106 | + Raises: |
| 107 | + InvalidGraphIoError: Outside a ``@pipeline`` body, on a duplicate |
| 108 | + name, on an unusable field, or on a constant value. |
| 109 | + """ |
| 110 | + builder = _active_builder("graph_output") |
| 111 | + if not when: |
| 112 | + return |
| 113 | + |
| 114 | + _check_name(name, "graph_output") |
| 115 | + if any(entry.get("name") == name for entry in builder.outputs): |
| 116 | + raise InvalidGraphIoError( |
| 117 | + f"graph output {name!r} is already declared on pipeline " |
| 118 | + f"{builder.name!r}. Output names must be unique." |
| 119 | + ) |
| 120 | + |
| 121 | + entry: dict[str, Any] = {"name": name} |
| 122 | + if type is not None: |
| 123 | + entry["type"] = _check_str("type", type) |
| 124 | + if description is not None: |
| 125 | + entry["description"] = _check_str("description", description) |
| 126 | + merged = _layout_annotations(annotations, position) |
| 127 | + if merged: |
| 128 | + entry["annotations"] = merged |
| 129 | + |
| 130 | + builder.outputs.append(entry) |
| 131 | + builder.output_values[name] = _edge(name, value) |
| 132 | + |
| 133 | + |
| 134 | +def _active_builder(caller: str) -> Any: |
| 135 | + from .trace import current_builder |
| 136 | + |
| 137 | + builder = current_builder() |
| 138 | + if builder is None: |
| 139 | + raise InvalidGraphIoError( |
| 140 | + f"{caller}() must be called inside a @pipeline function body, " |
| 141 | + "where a trace is active." |
| 142 | + ) |
| 143 | + return builder |
| 144 | + |
| 145 | + |
| 146 | +def _edge(name: str, value: Any) -> EdgeRef: |
| 147 | + if isinstance(value, TaskOutputProxy): |
| 148 | + return EdgeRef( |
| 149 | + kind="taskOutput", |
| 150 | + task_id=value._task_id, |
| 151 | + output=value._resolved_output_name(), |
| 152 | + ) |
| 153 | + if isinstance(value, GraphInputPlaceholder): |
| 154 | + return EdgeRef(kind="graphInput", input_name=value.input_name) |
| 155 | + raise InvalidGraphIoError( |
| 156 | + f"graph output {name!r} must be wired to a task output or a graph " |
| 157 | + f"input; got {type(value).__name__}. Constant graph outputs are not " |
| 158 | + "supported — emit the value from a task." |
| 159 | + ) |
| 160 | + |
| 161 | + |
| 162 | +def _layout_annotations( |
| 163 | + annotations: Mapping[str, str] | None, |
| 164 | + position: tuple[float, float] | None, |
| 165 | +) -> dict[str, str]: |
| 166 | + merged: dict[str, str] = {} |
| 167 | + if annotations is not None: |
| 168 | + if not isinstance(annotations, Mapping): |
| 169 | + raise InvalidGraphIoError( |
| 170 | + f"annotations must be a mapping; got {type(annotations).__name__}." |
| 171 | + ) |
| 172 | + check_annotations( |
| 173 | + annotations, |
| 174 | + policy=CALLER_ANNOTATION_POLICY, |
| 175 | + error_cls=InvalidGraphIoError, |
| 176 | + ) |
| 177 | + merged.update(annotations) |
| 178 | + if position is not None: |
| 179 | + # Applied after the mapping, matching ``@pipeline(flow_direction=...)``. |
| 180 | + if not isinstance(position, tuple) or len(position) != 2: |
| 181 | + raise InvalidEditorLayoutError( |
| 182 | + "position must be an (x, y) tuple; got " |
| 183 | + f"{type(position).__name__}." |
| 184 | + ) |
| 185 | + merged[POSITION_ANNOTATION] = position_annotation_value( |
| 186 | + *position, error_cls=InvalidEditorLayoutError |
| 187 | + ) |
| 188 | + return merged |
| 189 | + |
| 190 | + |
| 191 | +def _check_name(name: Any, caller: str) -> str: |
| 192 | + if not isinstance(name, str) or not name: |
| 193 | + raise InvalidGraphIoError( |
| 194 | + f"{caller}() name must be a non-empty string; got " |
| 195 | + f"{type(name).__name__}." |
| 196 | + ) |
| 197 | + return name |
| 198 | + |
| 199 | + |
| 200 | +def _check_str(field: str, value: Any, *, hint: str | None = None) -> str: |
| 201 | + if not isinstance(value, str): |
| 202 | + suffix = f". {hint.capitalize()}." if hint else "." |
| 203 | + raise InvalidGraphIoError( |
| 204 | + f"{field} must be a string; got {type(value).__name__}{suffix}" |
| 205 | + ) |
| 206 | + return value |
| 207 | + |
| 208 | + |
| 209 | +def _check_bool(value: Any) -> bool: |
| 210 | + if not isinstance(value, bool): |
| 211 | + raise InvalidGraphIoError( |
| 212 | + f"optional must be True or False; got {type(value).__name__}." |
| 213 | + ) |
| 214 | + return value |
0 commit comments