From ab7bc8a57c8a36f464deb3d448bb7af5026fe346 Mon Sep 17 00:00:00 2001 From: Volv G Date: Thu, 24 Sep 2026 22:05:44 -0700 Subject: [PATCH] Declare graph inputs and outputs from the pipeline body (v0.1.23) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pipeline's inputs come from its `In[T]` parameters and its outputs from the return annotation, which only covers graphs whose I/O is Python-shaped. A name that is not an identifier ("Pipeline Creation Time"), an exact Tangle type string, an input declared conditionally, or a per-input editor position has no spelling at all — which is why downstream code reaches into `current_builder().inputs` / `.output_values` behind private helpers. `graph_input()` / `graph_output()` declare the same entries on the active trace as a public, validated API: unique names checked against `In[...]` parameters and the return annotation, `default` implying `optional: true` the way a defaulted parameter already does, handles-not-constants for outputs, `when=` for conditional declaration, and `position=(x, y)` reusing the canonical editor.position serializer. Also fixes a latent bug: `graph_output()` beside a returned `Out[T]` was silently lossy, because emit prefers the multi-output map while the returned edge lived only in the legacy single-output shims. The returned output now joins the map, declared last. --- README.md | 33 +- .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../tangle_cli/python_pipeline/__init__.py | 3 + .../src/tangle_cli/python_pipeline/errors.py | 9 + .../tangle_cli/python_pipeline/graph_io.py | 214 +++++++ .../src/tangle_cli/python_pipeline/trace.py | 22 +- pyproject.toml | 2 +- tests/test_graph_io.py | 533 ++++++++++++++++++ tests/test_packaging.py | 2 +- tests/test_python_pipeline_dsl.py | 2 + uv.lock | 2 +- 11 files changed, 818 insertions(+), 6 deletions(-) create mode 100644 packages/tangle-cli/src/tangle_cli/python_pipeline/graph_io.py create mode 100644 tests/test_graph_io.py diff --git a/README.md b/README.md index 595025d..aa61685 100644 --- a/README.md +++ b/README.md @@ -603,6 +603,37 @@ The rules live in one place, `tangle_cli.schema_validation`: `check_annotations( A distribution that reads these annotations from its own config file should call `check_annotations(mapping, policy=CALLER_ANNOTATION_POLICY, error_cls=...)` at config-parse time — passing its own error type and adding the config path and key to the message — so one user mistake produces one diagnostic instead of two competing ones. The compiler's own call is then the backstop for anything arriving by another route. +##### Declaring graph inputs and outputs from the body + +A pipeline's inputs normally come from its `In[T]` parameters and its outputs from the return annotation. That only covers graphs whose I/O is Python-shaped. `graph_input()` / `graph_output()` declare the same entries directly on the active pipeline, for the shapes a signature cannot express: + +```python +from tangle_cli.python_pipeline import Out, graph_input, graph_output, pipeline + +@pipeline("Daily Pulse") +def daily_pulse() -> Out[str]: + created = graph_input( + "Pipeline Creation Time", "String", default="Use as a cache busting mechanism" + ) + limit = graph_input("query_limit", "Integer", optional=True, when=has_limit) + scrape = SCRAPE.named("Scrape").with_position(0, 0)(created=created, limit=limit) + graph_output("scrape_date", scrape.scrape_date, "String") + return scrape +``` + +- **A name is a string, not an identifier.** `"Pipeline Creation Time"` is a legal Tangle input name and an illegal Python parameter name; same for a type that is a Tangle type string (`"Json"`) rather than a Python type. +- **`when=False` declares nothing** and returns `None`, so a conditional input stays a single assignment instead of an `if`/`else` around the call site. +- **`default` implies `optional: true`** unless `optional=False` is passed. This matches what a defaulted `In[T]` parameter emits; the corpus also contains defaulted-but-required inputs, which is why the implication is overridable. +- **Declaration order is signature parameters first, then body order**, and it is the order the entries appear in the document. +- **A name is declared once.** Duplicates are rejected, including a collision with an `In[...]` parameter or with the name the return annotation contributes (`Out[T]`'s `output_name`, or an `Outputs` field). +- **Outputs wire to a handle, never a constant** — a task output or a graph input, the same rule the `Outputs` return path enforces. Returning a value *and* declaring outputs is supported: the returned output is appended after the declared ones. +- **Layout comes along**: `position=(x, y)` writes the same canonical `editor.position` annotation as `.with_position(...)`, which is how a graph input gets a position in the editor. `annotations={...}` is validated under the same caller policy as `pipeline_annotations`. +- **Validated up front, without echoing values.** A non-string `default` (the schema's `InputSpec.default` is a string), a non-string name/type/description, a non-boolean `optional`, a constant output, or a call outside a `@pipeline` body raises `InvalidGraphIoError` (a `CompileError`) naming the field, never the value. + +The signature route remains the recommended default: it is typed, it is checked by your IDE, and it reads better. Reach for these when porting existing YAML or when the shape genuinely needs it. + +Porting an existing helper that appended to the builder by hand (for example relevance-tools' `_gin` / `_gout`): the emitted document is identical except in two places. An input declared with a `default` and no explicit `optional` gains `optional: true`, because that is what a defaulted `In[T]` parameter already emits and what the majority of the corpus carries; pass `optional=False` to keep it required. An input declaring both keys emits `default` before `optional`, which is the order the corpus uses 28 times against 3 for the reverse. + ##### Editor layout (`editor.*`) The pipeline editor stores graph layout in ordinary annotations, so layout is authorable from Python. `.with_position(x, y)` on a task or subpipeline handle writes `editor.position`, and `@pipeline(flow_direction=...)` writes the root `editor.flow-direction`: @@ -630,7 +661,7 @@ SCRAPE.with_annotations({"editor.position": '{"x": 0, "y": 0}'}) - **Positions are descriptive.** They are task annotations, not part of any `componentRef`, so they never change component digests, compile identity, or cache behaviour. On a subpipeline handle the position applies to the parent task; the child sidecar is byte-identical either way. - **Auto-layout interaction.** At submit time the runner lays a graph out only when no task carries a *non-zero* position, so an explicit `.with_position(...)` suppresses auto-layout and `--force-layout` overrides that. A graph positioned entirely at `(0, 0)` still counts as unpositioned and is laid out. `tangle sdk pipelines layout` writes the same canonical value this sugar writes. -Graph **inputs** and **outputs** carry `editor.position` in the same way, and the editor reads it there, but there is no Python authoring surface for a graph input/output object yet — inputs come from the `In[T]` signature and outputs from the return annotation, neither of which has a place to hang layout. That is follow-up work tied to a public `graph_input()` / `graph_output()` API; until then, position inputs by editing the YAML or by using the editor. +Graph **inputs** and **outputs** carry `editor.position` in the same way. An `In[T]` parameter has nowhere to hang layout, so positioning those means declaring them with `graph_input(..., position=(x, y))` instead — see the section above. The key names, the serialized format, and the validation live in `tangle_cli.editor_layout` (`POSITION_ANNOTATION`, `FLOW_DIRECTION_ANNOTATION`, `FLOW_DIRECTIONS`, `position_annotation_value`, `validate_flow_direction`), which the CLI's own `pipelines layout` and the runner's auto-layout gate share, so a tool emitting layout cannot drift from what the editor reads. That module is stdlib-only and lives outside `python_pipeline` on purpose, so layout and submit paths do not pull in the authoring/codegen stack; it raises whatever `error_cls` the caller injects, and the authoring surfaces inject `InvalidEditorLayoutError`. diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index 69f5486..540ae8a 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.22" + __version__ = "0.1.23" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py index 56bf560..b432643 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py @@ -21,6 +21,7 @@ from __future__ import annotations from .dynamic_data import dynamic_secret +from .graph_io import graph_input, graph_output from .pipeline import pipeline from .raw import raw from .ref import ref @@ -33,6 +34,8 @@ __all__ = [ "pipeline", + "graph_input", + "graph_output", "task", "Publish", "registered", diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/errors.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/errors.py index 354c42d..924f4e2 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/errors.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/errors.py @@ -38,6 +38,15 @@ class InvalidEditorLayoutError(CompileError): """ +class InvalidGraphIoError(CompileError): + """Raised on an unusable ``graph_input()`` / ``graph_output()`` call. + + Covers declaration outside a trace, duplicate names, fields the pipeline + schema would reject, and constant graph outputs. Messages name the field, + never the value. + """ + + class InvalidPipelineAnnotationsError(CompileError): """Raised on a malformed caller-supplied ``pipeline_annotations`` mapping. diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/graph_io.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/graph_io.py new file mode 100644 index 0000000..cc5d504 --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/graph_io.py @@ -0,0 +1,214 @@ +"""``graph_input()`` / ``graph_output()`` — declare graph I/O from the body. + +A pipeline's inputs normally come from its ``In[T]`` parameters and its +outputs from the return annotation. That covers Python-shaped graphs only: +it cannot express a name that is not an identifier (``"Pipeline Creation +Time"``), an exact Tangle type string, an input declared conditionally, or +a per-input ``editor.position``. These two functions declare the same +entries directly on the active trace, so a YAML pipeline that uses those +shapes stays expressible in Python. + +Entry key order follows the tracer's signature path and the corpus: +``name, type, description, default, optional, annotations``. +""" +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from tangle_cli.editor_layout import POSITION_ANNOTATION, position_annotation_value +from tangle_cli.schema_validation import CALLER_ANNOTATION_POLICY, check_annotations + +from .errors import InvalidEditorLayoutError, InvalidGraphIoError +from .graph import EdgeRef +from .placeholders import GraphInputPlaceholder, TaskOutputProxy + +# Distinguishes "no default" from an explicit ``default=None``, which is a +# value the schema does not accept and must be reported, not dropped. +_UNSET: Any = object() + +__all__ = ["graph_input", "graph_output"] + + +def graph_input( + name: str, + type: str | None = None, + *, + description: str | None = None, + default: Any = _UNSET, + optional: bool | None = None, + annotations: Mapping[str, str] | None = None, + position: tuple[float, float] | None = None, + when: bool = True, +) -> GraphInputPlaceholder | None: + """Declare a graph input on the active pipeline and return its handle. + + The handle is passed to task arguments exactly like an ``In[T]`` + parameter. Supplying ``default`` implies ``optional: true`` unless + ``optional=False`` is passed explicitly. + + ``when=False`` declares nothing and returns ``None``, so a conditional + input reads as one assignment instead of an ``if``/``else``. + + Raises: + InvalidGraphIoError: Outside a ``@pipeline`` body, on a duplicate + name, or on an unusable field. Messages name the field, never + the value. + """ + builder = _active_builder("graph_input") + if not when: + return None + + _check_name(name, "graph_input") + if any(entry.get("name") == name for entry in builder.inputs): + # Also catches an ``In[...]`` parameter: signature inputs are + # appended before the body runs. + raise InvalidGraphIoError( + f"graph input {name!r} is already declared on pipeline " + f"{builder.name!r}. Input names must be unique." + ) + + entry: dict[str, Any] = {"name": name} + if type is not None: + entry["type"] = _check_str("type", type) + if description is not None: + entry["description"] = _check_str("description", description) + if default is not _UNSET: + # InputSpec.default is a string in the schema; a non-string default + # would only fail later, against a schema path instead of this call. + entry["default"] = _check_str("default", default, hint="pass str(value)") + entry["optional"] = True if optional is None else _check_bool(optional) + elif optional is not None: + entry["optional"] = _check_bool(optional) + merged = _layout_annotations(annotations, position) + if merged: + entry["annotations"] = merged + + builder.inputs.append(entry) + return GraphInputPlaceholder(input_name=name) + + +def graph_output( + name: str, + value: Any, + type: str | None = None, + *, + description: str | None = None, + annotations: Mapping[str, str] | None = None, + position: tuple[float, float] | None = None, + when: bool = True, +) -> None: + """Declare a graph output on the active pipeline, wired to ``value``. + + ``value`` is a task output handle or a graph-input handle; a constant is + refused, matching the ``Outputs`` return path. + + Raises: + InvalidGraphIoError: Outside a ``@pipeline`` body, on a duplicate + name, on an unusable field, or on a constant value. + """ + builder = _active_builder("graph_output") + if not when: + return + + _check_name(name, "graph_output") + if any(entry.get("name") == name for entry in builder.outputs): + raise InvalidGraphIoError( + f"graph output {name!r} is already declared on pipeline " + f"{builder.name!r}. Output names must be unique." + ) + + entry: dict[str, Any] = {"name": name} + if type is not None: + entry["type"] = _check_str("type", type) + if description is not None: + entry["description"] = _check_str("description", description) + merged = _layout_annotations(annotations, position) + if merged: + entry["annotations"] = merged + + builder.outputs.append(entry) + builder.output_values[name] = _edge(name, value) + + +def _active_builder(caller: str) -> Any: + from .trace import current_builder + + builder = current_builder() + if builder is None: + raise InvalidGraphIoError( + f"{caller}() must be called inside a @pipeline function body, " + "where a trace is active." + ) + return builder + + +def _edge(name: str, value: Any) -> EdgeRef: + if isinstance(value, TaskOutputProxy): + return EdgeRef( + kind="taskOutput", + task_id=value._task_id, + output=value._resolved_output_name(), + ) + if isinstance(value, GraphInputPlaceholder): + return EdgeRef(kind="graphInput", input_name=value.input_name) + raise InvalidGraphIoError( + f"graph output {name!r} must be wired to a task output or a graph " + f"input; got {type(value).__name__}. Constant graph outputs are not " + "supported — emit the value from a task." + ) + + +def _layout_annotations( + annotations: Mapping[str, str] | None, + position: tuple[float, float] | None, +) -> dict[str, str]: + merged: dict[str, str] = {} + if annotations is not None: + if not isinstance(annotations, Mapping): + raise InvalidGraphIoError( + f"annotations must be a mapping; got {type(annotations).__name__}." + ) + check_annotations( + annotations, + policy=CALLER_ANNOTATION_POLICY, + error_cls=InvalidGraphIoError, + ) + merged.update(annotations) + if position is not None: + # Applied after the mapping, matching ``@pipeline(flow_direction=...)``. + if not isinstance(position, tuple) or len(position) != 2: + raise InvalidEditorLayoutError( + "position must be an (x, y) tuple; got " + f"{type(position).__name__}." + ) + merged[POSITION_ANNOTATION] = position_annotation_value( + *position, error_cls=InvalidEditorLayoutError + ) + return merged + + +def _check_name(name: Any, caller: str) -> str: + if not isinstance(name, str) or not name: + raise InvalidGraphIoError( + f"{caller}() name must be a non-empty string; got " + f"{type(name).__name__}." + ) + return name + + +def _check_str(field: str, value: Any, *, hint: str | None = None) -> str: + if not isinstance(value, str): + suffix = f". {hint.capitalize()}." if hint else "." + raise InvalidGraphIoError( + f"{field} must be a string; got {type(value).__name__}{suffix}" + ) + return value + + +def _check_bool(value: Any) -> bool: + if not isinstance(value, bool): + raise InvalidGraphIoError( + f"optional must be True or False; got {type(value).__name__}." + ) + return value diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/trace.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/trace.py index 1f46b87..6f955dd 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/trace.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/trace.py @@ -285,13 +285,20 @@ def trace_pipeline( inner = _annotation_inner_type(return_anno) type_str = _python_type_to_tangle_type(inner) output_name = pipeline_fn.output_name + _ensure_no_duplicate_output(builder, output_name) builder.outputs.append({"name": output_name, "type": type_str}) builder.output_name = output_name - builder.output_taskref = EdgeRef( + edge = EdgeRef( kind="taskOutput", task_id=result._task_id, output=result._resolved_output_name(), ) + builder.output_taskref = edge + if builder.output_values: + # graph_output() already populated the multi-output map, which + # emit PREFERS; the returned output has to join it or it would be + # silently dropped. Declared last, after the body's own outputs. + builder.output_values[output_name] = edge elif return_anno is not inspect.Signature.empty and _is_outputs_class(return_anno): _trace_multi_output(builder, pipeline_fn, return_anno, result) @@ -355,9 +362,22 @@ def _trace_multi_output( "outputs." ) type_str = _python_type_to_tangle_type(inner) + _ensure_no_duplicate_output(builder, field_name) builder.outputs.append({"name": field_name, "type": type_str}) builder.output_values[field_name] = edge +def _ensure_no_duplicate_output(builder: GraphBuilder, name: str) -> None: + """Reject a return-annotation output that ``graph_output()`` already + declared, instead of emitting the name twice.""" + if any(entry.get("name") == name for entry in builder.outputs): + from .errors import InvalidGraphIoError + + raise InvalidGraphIoError( + f"graph output {name!r} is declared by graph_output() and by the " + f"return annotation of pipeline {builder.name!r}. Declare it once." + ) + + # Re-export commonly-used names. __all__ = ["current_builder", "trace_pipeline"] diff --git a/pyproject.toml b/pyproject.toml index 53f9a69..6607490 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.22" +version = "0.1.23" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_graph_io.py b/tests/test_graph_io.py new file mode 100644 index 0000000..366bce1 --- /dev/null +++ b/tests/test_graph_io.py @@ -0,0 +1,533 @@ +"""``graph_input()`` / ``graph_output()`` — graph I/O declared from the body. + +The signature/return-annotation route covers Python-shaped graphs only. These +functions exist for the shapes it cannot express, and the contract pinned here +is what a YAML→Python port needs: + +* non-identifier names and exact Tangle type strings survive to the YAML; +* declaration order is signature parameters first, then body order; +* ``default`` implies ``optional: true`` unless ``optional=False``; +* a name may be declared once — including against ``In[...]`` parameters and + the return annotation; +* constants are not graph outputs; +* diagnostics name the field, never the value. +""" + +from __future__ import annotations + +import textwrap +from pathlib import Path + +import pytest +import yaml + +from tangle_cli.pipeline_compiler import compile_pipeline +from tangle_cli.python_pipeline import graph_input, graph_output +from tangle_cli.python_pipeline.errors import ( + CompileError, + InvalidEditorLayoutError, + InvalidGraphIoError, +) + +_HEADER = ''' +from dataclasses import dataclass + +from tangle_cli.python_pipeline import ( + In, + Out, + Outputs, + graph_input, + graph_output, + pipeline, + subpipeline, + task, +) + + +@task(image="python:3.12") +def greet(greeting: str = "hi", limit: str = "0"): + """Write a greeting. + + Metadata: + Name: Greet + """ + print(greeting, limit) + +''' + + +def _compile(tmp_path: Path, body: str, case: str, *, pipeline_name=None) -> Path: + """Compile ``_HEADER + body`` in its own directory; return the output.""" + case_dir = tmp_path / case + case_dir.mkdir(parents=True, exist_ok=True) + script = case_dir / "pipeline.py" + script.write_text(_HEADER + textwrap.dedent(body), encoding="utf-8") + out = case_dir / "compiled.yaml" + compile_pipeline(script, out, pipeline_name=pipeline_name) + return out + + +def _doc(out: Path) -> dict: + return yaml.safe_load(out.read_text(encoding="utf-8")) + + +def _expect(tmp_path, body, case, error=InvalidGraphIoError, **kwargs): + with pytest.raises(error) as exc: + _compile(tmp_path, body, case, **kwargs) + return str(exc.value) + + +# --------------------------------------------------------------------------- +# What the signature cannot express. + + +def test_a_non_identifier_name_and_exact_type_reach_the_document(tmp_path): + """The reason this API exists: ``"Pipeline Creation Time"`` is not a legal + parameter name, and ``Json`` is not a Python type.""" + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + created = graph_input( + "Pipeline Creation Time", + "String", + default="Use as a cache busting mechanism", + ) + params = graph_input("template_params_override", "Json", optional=True) + run_greet = greet(greeting=created, limit=params) + return run_greet + ''', + "shapes", + ) + + assert _doc(out)["inputs"] == [ + { + "name": "Pipeline Creation Time", + "type": "String", + "default": "Use as a cache busting mechanism", + "optional": True, + }, + {"name": "template_params_override", "type": "Json", "optional": True}, + ] + + +def test_signature_inputs_come_first_then_declaration_order(tmp_path): + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p(from_signature: In[str] = "x") -> Out[str]: + second = graph_input("second", "String") + third = graph_input("third", "String") + run_greet = greet(greeting=second, limit=third) + return run_greet + ''', + "order", + ) + + assert [i["name"] for i in _doc(out)["inputs"]] == [ + "from_signature", + "second", + "third", + ] + + +def test_a_conditional_input_declares_nothing_when_false(tmp_path): + """``when=False`` keeps a conditional declaration a single assignment.""" + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + absent = graph_input("absent", "String", when=False) + assert absent is None + run_greet = greet() + graph_output("skipped", run_greet, "String", when=False) + return run_greet + ''', + "when", + ) + + doc = _doc(out) + assert "inputs" not in doc + assert [o["name"] for o in doc["outputs"]] == ["wait_for_output"] + + +# --------------------------------------------------------------------------- +# optional / default. + + +def test_a_default_implies_optional_and_matches_the_signature_route(tmp_path): + """A defaulted ``In[T]`` parameter emits ``default`` + ``optional: true``; + the explicit route emits the same entry for the same shape.""" + declared = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + value = graph_input("value", "String", default="x") + run_greet = greet(greeting=value) + return run_greet + ''', + "declared", + ) + from_signature = _compile( + tmp_path, + ''' + @pipeline("P") + def p(value: In[str] = "x") -> Out[str]: + run_greet = greet(greeting=value) + return run_greet + ''', + "signature", + ) + + assert _doc(declared)["inputs"] == _doc(from_signature)["inputs"] + + +def test_an_explicit_optional_false_survives_a_default(tmp_path): + """The corpus contains defaulted-but-required inputs, so the implication + has to be overridable.""" + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + value = graph_input("value", "String", default="x", optional=False) + run_greet = greet(greeting=value) + return run_greet + ''', + "required_default", + ) + + assert _doc(out)["inputs"] == [ + {"name": "value", "type": "String", "default": "x", "optional": False} + ] + + +def test_optional_alone_emits_no_default(tmp_path): + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + value = graph_input("value", "String", optional=True) + run_greet = greet(greeting=value) + return run_greet + ''', + "optional_only", + ) + + assert _doc(out)["inputs"] == [ + {"name": "value", "type": "String", "optional": True} + ] + + +def test_a_non_string_default_is_refused_without_echoing_it(tmp_path): + """``InputSpec.default`` is a string in the pipeline schema; catching it + here names the field instead of failing later against a schema path.""" + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + value = graph_input("value", "Integer", default=41) + run_greet = greet(greeting=value) + return run_greet + ''', + "bad_default", + ) + + assert "default" in message and "int" in message + assert "41" not in message + + +# --------------------------------------------------------------------------- +# Uniqueness. + + +def test_a_duplicate_input_name_is_refused(tmp_path): + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + first = graph_input("value", "String") + second = graph_input("value", "String") + run_greet = greet(greeting=first, limit=second) + return run_greet + ''', + "dup_input", + ) + + assert "'value'" in message + + +def test_an_input_colliding_with_a_signature_parameter_is_refused(tmp_path): + """Signature inputs are appended before the body runs, so the collision is + visible at declaration time.""" + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p(value: In[str] = "x") -> Out[str]: + shadow = graph_input("value", "String") + run_greet = greet(greeting=shadow) + return run_greet + ''', + "dup_signature", + ) + + assert "'value'" in message + + +def test_a_duplicate_output_name_is_refused(tmp_path): + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + run_greet = greet() + graph_output("result", run_greet.a, "String") + graph_output("result", run_greet.b, "String") + return run_greet + ''', + "dup_output", + ) + + assert "'result'" in message + + +def test_an_output_colliding_with_the_return_annotation_is_refused(tmp_path): + """The return output is added after the body, so this is caught there.""" + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + run_greet = greet() + graph_output("wait_for_output", run_greet.rows, "String") + return run_greet + ''', + "dup_return", + ) + + assert "'wait_for_output'" in message + + +def test_an_output_colliding_with_an_outputs_field_is_refused(tmp_path): + message = _expect( + tmp_path, + ''' + @dataclass(frozen=True) + class Result(Outputs): + rows: Out[str] + + + @pipeline("P") + def p() -> Result: + run_greet = greet() + graph_output("rows", run_greet.rows, "String") + return Result(rows=run_greet.rows) + ''', + "dup_outputs_field", + ) + + assert "'rows'" in message + + +# --------------------------------------------------------------------------- +# Outputs and wiring. + + +def test_outputs_wire_from_a_task_output_or_a_graph_input(tmp_path): + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + created = graph_input("created", "String") + run_greet = greet(greeting=created) + graph_output("scrape_date", run_greet.scrape_date, "String") + graph_output("passthrough", created, "String") + return run_greet + ''', + "wiring", + ) + + values = _doc(out)["implementation"]["graph"]["outputValues"] + assert values["scrape_date"] == { + "taskOutput": {"taskId": "Run Greet", "outputName": "scrape_date"} + } + assert values["passthrough"] == {"graphInput": {"inputName": "created"}} + + +def test_the_returned_output_is_kept_alongside_declared_outputs(tmp_path): + """Regression: emit PREFERS the multi-output map, so a declared output + must not displace the value the pipeline returns.""" + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + run_greet = greet() + graph_output("scrape_date", run_greet.scrape_date, "String") + return run_greet + ''', + "return_kept", + ) + + doc = _doc(out) + assert [o["name"] for o in doc["outputs"]] == ["scrape_date", "wait_for_output"] + assert list(doc["implementation"]["graph"]["outputValues"]) == [ + "scrape_date", + "wait_for_output", + ] + + +def test_a_constant_graph_output_is_refused_without_echoing_it(tmp_path): + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + run_greet = greet() + graph_output("literal", "s3://bucket/secret-path", "String") + return run_greet + ''', + "constant_output", + ) + + assert "str" in message + assert "s3://bucket/secret-path" not in message + + +# --------------------------------------------------------------------------- +# Layout sugar reuse. + + +def test_position_and_annotations_are_written_to_the_entry(tmp_path): + out = _compile( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + created = graph_input( + "created", + "String", + annotations={"sdk": "x"}, + position=(-540, 860), + ) + run_greet = greet(greeting=created) + graph_output("done", run_greet, "String", position=(120, 40)) + return run_greet + ''', + "layout", + ) + + doc = _doc(out) + assert doc["inputs"][0]["annotations"] == { + "sdk": "x", + "editor.position": '{"x": -540, "y": 860}', + } + assert doc["outputs"][0]["annotations"] == { + "editor.position": '{"x": 120, "y": 40}' + } + + +def test_a_bad_position_raises_the_layout_error(tmp_path): + _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + created = graph_input("created", "String", position=(float("nan"), 0)) + run_greet = greet(greeting=created) + return run_greet + ''', + "bad_position", + error=InvalidEditorLayoutError, + ) + + +def test_a_rejected_annotation_value_is_not_echoed(tmp_path): + """Annotations go through the same caller policy as + ``pipeline_annotations``, so a config-sourced value is never echoed.""" + message = _expect( + tmp_path, + ''' + @pipeline("P") + def p() -> Out[str]: + created = graph_input( + "created", "String", annotations={"token": ["super-secret"]} + ) + run_greet = greet(greeting=created) + return run_greet + ''', + "bad_annotation", + ) + + assert "'token'" in message and "list" in message + assert "super-secret" not in message + + +# --------------------------------------------------------------------------- +# Subpipelines. + + +def test_a_child_declared_input_is_bound_from_the_parent_call_site(tmp_path): + """A child's declared inputs are ordinary graph inputs, so the parent + passes them by name exactly as it would an ``In[T]`` parameter.""" + out = _compile( + tmp_path, + ''' + @pipeline("Child") + def child() -> Out[str]: + created = graph_input("Pipeline Creation Time", "String") + run_greet = greet(greeting=created) + return run_greet + + + @pipeline("Parent") + def parent() -> Out[str]: + parent_value = graph_input("parent_value", "String") + run_child = subpipeline(child)( + **{"Pipeline Creation Time": parent_value} + ) + return run_child + ''', + "subpipeline", + pipeline_name="parent", + ) + + parent_task = _doc(out)["implementation"]["graph"]["tasks"]["Run Child"] + assert parent_task["arguments"] == { + "Pipeline Creation Time": {"graphInput": {"inputName": "parent_value"}} + } + child_path = next( + p for p in out.parent.rglob("child-*.yaml") if ".components" not in p.name + ) + child_doc = yaml.safe_load(child_path.read_text(encoding="utf-8")) + assert child_doc["inputs"] == [ + {"name": "Pipeline Creation Time", "type": "String"} + ] + + +# --------------------------------------------------------------------------- +# Misuse outside a trace. + + +def test_declaring_outside_a_pipeline_body_is_refused(): + """There is no active graph to declare on, and a silent no-op would lose + the declaration.""" + with pytest.raises(InvalidGraphIoError) as exc: + graph_input("value", "String") + assert "@pipeline" in str(exc.value) + + with pytest.raises(InvalidGraphIoError): + graph_output("value", None, "String") + + +def test_graph_io_errors_are_compile_errors(): + assert issubclass(InvalidGraphIoError, CompileError) diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 93dc39c..5a5635f 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.22" in metadata + assert "Version: 0.1.23" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_python_pipeline_dsl.py b/tests/test_python_pipeline_dsl.py index f4211db..d2e826c 100644 --- a/tests/test_python_pipeline_dsl.py +++ b/tests/test_python_pipeline_dsl.py @@ -52,6 +52,8 @@ class TestPublicSurface: def test_all_names_are_exported(self): assert set(pp.__all__) == { "pipeline", + "graph_input", + "graph_output", "task", "Publish", "registered", diff --git a/uv.lock b/uv.lock index 7dc3fc4..02cfbe1 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.22" +version = "0.1.23" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },